<!DOCTYPE html>
<html>
<head><meta name="generator" content="Hexo 3.9.0">
  <meta charset="utf-8">
  
  <title>元数据平台-数据血缘 | 月迹的博客</title>
  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
  <meta name="description" content="架构 首先，我们先了解一下元数据平台架构，主流程是：SQL采集 –》 SQL解析 –》应用层。  SQL采集：针对各种SQL查询引擎，编写相应的钩子函数进行SQL收集，收集内容有执行时间、执行耗时、执行用户、执行引擎、jobId和执行SQL等等，最后把SQL信息入Kafka。Hive是实现ExecuteWithHookContext接口，Presto是实现EventListener接口，Spark">
<meta name="keywords" content="大数据,hive,数据血缘,SQL采集,SQL解析">
<meta property="og:type" content="article">
<meta property="og:title" content="元数据平台-数据血缘">
<meta property="og:url" content="https://xyueji.github.io/元数据平台-数据血缘.html">
<meta property="og:site_name" content="月迹的博客">
<meta property="og:description" content="架构 首先，我们先了解一下元数据平台架构，主流程是：SQL采集 –》 SQL解析 –》应用层。  SQL采集：针对各种SQL查询引擎，编写相应的钩子函数进行SQL收集，收集内容有执行时间、执行耗时、执行用户、执行引擎、jobId和执行SQL等等，最后把SQL信息入Kafka。Hive是实现ExecuteWithHookContext接口，Presto是实现EventListener接口，Spark">
<meta property="og:locale" content="zh-CN">
<meta property="og:image" content="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210903154320404.png">
<meta property="og:image" content="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210903170253930.png">
<meta property="og:image" content="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210902172517522.png">
<meta property="og:updated_time" content="2021-09-23T09:17:14.580Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="元数据平台-数据血缘">
<meta name="twitter:description" content="架构 首先，我们先了解一下元数据平台架构，主流程是：SQL采集 –》 SQL解析 –》应用层。  SQL采集：针对各种SQL查询引擎，编写相应的钩子函数进行SQL收集，收集内容有执行时间、执行耗时、执行用户、执行引擎、jobId和执行SQL等等，最后把SQL信息入Kafka。Hive是实现ExecuteWithHookContext接口，Presto是实现EventListener接口，Spark">
<meta name="twitter:image" content="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210903154320404.png">
  
    <link rel="alternative" href="/atom.xml" title="月迹的博客" type="application/atom+xml">
  
  
    <link rel="icon" href="/img/favicon.png">
  
  
      <link rel="stylesheet" href="//cdn.bootcss.com/animate.css/3.5.0/animate.min.css">
  
  <link rel="stylesheet" href="css/style.css">
  <link rel="stylesheet" href="/font-awesome/css/font-awesome.min.css">
  <link rel="apple-touch-icon" href="/apple-touch-icon.png">
  
  
      <link rel="stylesheet" href="/fancybox/jquery.fancybox.css">
  
    
  <!-- 加载特效 -->
    <script src="/js/pace.js"></script>
    <link href="/css/pace/pace-theme-flash.css" rel="stylesheet">
  <script>
      var yiliaConfig = {
          rootUrl: '/',
          fancybox: true,
          animate: true,
          isHome: false,
          isPost: true,
          isArchive: false,
          isTag: false,
          isCategory: false,
          open_in_new: false
      }
  </script>
    <script src="https://cdn.bootcdn.net/ajax/libs/gitalk/1.6.2/gitalk.min.js"></script>
    <script>
        (function(){
            var bp = document.createElement('script');
            var curProtocol = window.location.protocol.split(':')[0];
            if (curProtocol === 'https') {
                bp.src = 'https://zz.bdstatic.com/linksubmit/push.js';
            }
            else {
                bp.src = 'http://push.zhanzhang.baidu.com/push.js';
            }
            var s = document.getElementsByTagName("script")[0];
            s.parentNode.insertBefore(bp, s);
        })();
    </script>
</head>
</html>
<body>
  <div id="container">
    <div class="left-col">
    <div class="overlay"></div>
<div class="intrude-less">
    <header id="header" class="inner">
        <a href="/" class="profilepic">
            
            <img lazy-src="/img/head.png" class="js-avatar">
            
        </a>

        <hgroup>
          <h1 class="header-author"><a href="/" title="Hi Mate">月 迹</a></h1>
        </hgroup>

        
        <p class="header-subtitle">用代码书写人生</p>
        
        
            <input type="text" class="search" id="search" style="text-align: center" placeholder="搜点什么...">
        
        
            <div id="switch-btn" class="switch-btn">
                <div class="icon">
                    <div class="icon-ctn">
                        <div class="icon-wrap icon-house" data-idx="0">
                            <div class="birdhouse"></div>
                            <div class="birdhouse_holes"></div>
                        </div>
                        <div class="icon-wrap icon-ribbon hide" data-idx="1">
                            <div class="ribbon"></div>
                        </div>
                        
                        <div class="icon-wrap icon-link hide" data-idx="2">
                            <div class="loopback_l"></div>
                            <div class="loopback_r"></div>
                        </div>
                        
                        
                        <div class="icon-wrap icon-me hide" data-idx="3">
                            <div class="user"></div>
                            <div class="shoulder"></div>
                        </div>
                        
                    </div>

                </div>
                <div class="tips-box hide">
                    <div class="tips-arrow"></div>
                    <ul class="tips-inner">
                        <li>菜单</li>
                        <li>标签</li>
                        
                        <li>友情链接</li>
                        
                        
                        <li>关于我</li>
                        
                    </ul>
                </div>
            </div>
        

        <div id="switch-area" class="switch-area">
            <div class="switch-wrap">
                <section class="switch-part switch-part1">
                    <nav class="header-menu">
                        <ul>
                        
                            <li><a href="index.html">博客首页</a></li>
                        
                            <li><a href="/archives">文章归档</a></li>
                        
                            <li><a href="/tags">文章统计</a></li>
                        
                            <li><a href="/message">交友留言</a></li>
                        
                            <li><a href="/navigation">资源导航</a></li>
                        
                            <li><a href="/about">关于作者</a></li>
                        
                        </ul>
                    </nav>
                    <nav class="header-nav">
                        <ul class="social">
                            
                                <a class="fl mail" target="_blank" href="http://mail.qq.com/cgi-bin/qm_share?t=qm_mailme&email=U2FhYmdrZGVjamITIiJ9MDw_" title="mail">mail</a>
                            
                                <a class="fl github" target="_blank" href="https://github.com/xyueji" title="github">github</a>
                            
                                <a class="fl rss" target="_blank" href="/atom.xml" title="rss">rss</a>
                            
                        </ul>
                    </nav>
                </section>

                
                <section class="switch-part switch-part2">
                    <div class="widget tagcloud" id="js-tagcloud">
                        <a href="tags/AppendTrie树/" style="font-size: 10px;">AppendTrie树</a> <a href="tags/HDFS/" style="font-size: 10px;">HDFS</a> <a href="tags/JanusGraph/" style="font-size: 10px;">JanusGraph</a> <a href="tags/SQL解析/" style="font-size: 10px;">SQL解析</a> <a href="tags/SQL采集/" style="font-size: 10px;">SQL采集</a> <a href="tags/alluxio/" style="font-size: 10px;">alluxio</a> <a href="tags/bitmap/" style="font-size: 10px;">bitmap</a> <a href="tags/container/" style="font-size: 10px;">container</a> <a href="tags/datasource/" style="font-size: 10px;">datasource</a> <a href="tags/docker/" style="font-size: 15px;">docker</a> <a href="tags/docker-compose/" style="font-size: 15px;">docker-compose</a> <a href="tags/flink/" style="font-size: 15px;">flink</a> <a href="tags/free/" style="font-size: 10px;">free</a> <a href="tags/git/" style="font-size: 10px;">git</a> <a href="tags/grafana/" style="font-size: 10px;">grafana</a> <a href="tags/hadoop/" style="font-size: 10px;">hadoop</a> <a href="tags/hbase/" style="font-size: 15px;">hbase</a> <a href="tags/hbck/" style="font-size: 10px;">hbck</a> <a href="tags/hive/" style="font-size: 15px;">hive</a> <a href="tags/idea/" style="font-size: 15px;">idea</a> <a href="tags/impala/" style="font-size: 10px;">impala</a> <a href="tags/iptables/" style="font-size: 10px;">iptables</a> <a href="tags/mac/" style="font-size: 10px;">mac</a> <a href="tags/mapreduce/" style="font-size: 10px;">mapreduce</a> <a href="tags/metrics2/" style="font-size: 10px;">metrics2</a> <a href="tags/opentsdb/" style="font-size: 10px;">opentsdb</a> <a href="tags/phoenix/" style="font-size: 15px;">phoenix</a> <a href="tags/presto/" style="font-size: 10px;">presto</a> <a href="tags/prometheus/" style="font-size: 10px;">prometheus</a> <a href="tags/pycharm/" style="font-size: 10px;">pycharm</a> <a href="tags/salt-api/" style="font-size: 10px;">salt-api</a> <a href="tags/shuffle/" style="font-size: 15px;">shuffle</a> <a href="tags/spark/" style="font-size: 15px;">spark</a> <a href="tags/springboot/" style="font-size: 10px;">springboot</a> <a href="tags/top/" style="font-size: 10px;">top</a> <a href="tags/vue/" style="font-size: 10px;">vue</a> <a href="tags/wordcount/" style="font-size: 10px;">wordcount</a> <a href="tags/修复工具/" style="font-size: 10px;">修复工具</a> <a href="tags/元数据/" style="font-size: 10px;">元数据</a> <a href="tags/分片/" style="font-size: 10px;">分片</a> <a href="tags/基础知识/" style="font-size: 10px;">基础知识</a> <a href="tags/大数据/" style="font-size: 20px;">大数据</a> <a href="tags/大文件/" style="font-size: 10px;">大文件</a> <a href="tags/大查询/" style="font-size: 10px;">大查询</a> <a href="tags/安装/" style="font-size: 10px;">安装</a> <a href="tags/小文件/" style="font-size: 10px;">小文件</a> <a href="tags/开发规范/" style="font-size: 10px;">开发规范</a> <a href="tags/数据倾斜/" style="font-size: 10px;">数据倾斜</a> <a href="tags/数据血缘/" style="font-size: 10px;">数据血缘</a> <a href="tags/测试/" style="font-size: 10px;">测试</a> <a href="tags/热门商品/" style="font-size: 10px;">热门商品</a> <a href="tags/端口/" style="font-size: 10px;">端口</a> <a href="tags/端口映射/" style="font-size: 10px;">端口映射</a> <a href="tags/精准去重/" style="font-size: 10px;">精准去重</a> <a href="tags/系统表/" style="font-size: 10px;">系统表</a> <a href="tags/表重建/" style="font-size: 10px;">表重建</a>
                    </div>
                </section>
                

                
                <section class="switch-part switch-part3">
                    <div id="js-friends">
                    
                      <a target="_blank" class="main-nav-link switch-friends-link" href="https://dayesahh.github.io/">DaYesahh</a>
                    
                      <a target="_blank" class="main-nav-link switch-friends-link" href="http://www.54tianzhisheng.cn/">zhisjeng</a>
                    
                    </div>
                </section>
                

                
                
                <section class="switch-part switch-part4">
                
                    <div id="js-aboutme">爱动漫、追小说、喜影视、迷于严谨和逻辑的编码之中！！！</div>
                </section>
                
            </div>
        </div>
    </header>
</div>

    </div>
    <div class="mid-col">
      <nav id="mobile-nav">
      <div class="overlay">
          <div class="slider-trigger"></div>
          <h1 class="header-author js-mobile-header hide"><a href="/" title="Me">月 迹</a></h1>
      </div>
    <div class="intrude-less">
        <header id="header" class="inner">
            <a href="/" class="profilepic">
                
                    <img lazy-src="/img/head.png" class="js-avatar">
                
            </a>
            <hgroup>
              <h1 class="header-author"><a href="/" title="Me">月 迹</a></h1>
            </hgroup>
            
            <p class="header-subtitle">用代码书写人生</p>
            
            <nav class="header-menu">
                <ul>
                
                    <li><a href="index.html">博客首页</a></li>
                
                    <li><a href="/archives">文章归档</a></li>
                
                    <li><a href="/tags">文章统计</a></li>
                
                    <li><a href="/message">交友留言</a></li>
                
                    <li><a href="/navigation">资源导航</a></li>
                
                    <li><a href="/about">关于作者</a></li>
                
                <div class="clearfix"></div>
                </ul>
            </nav>
            <nav class="header-nav">
                <div class="social">
                    
                        <a class="mail" target="_blank" href="http://mail.qq.com/cgi-bin/qm_share?t=qm_mailme&email=U2FhYmdrZGVjamITIiJ9MDw_" title="mail">mail</a>
                    
                        <a class="github" target="_blank" href="https://github.com/xyueji" title="github">github</a>
                    
                        <a class="rss" target="_blank" href="/atom.xml" title="rss">rss</a>
                    
                </div>
            </nav>
        </header>                
    </div>
</nav>
      <div class="body-wrap"><article id="post-元数据平台-数据血缘" class="article article-type-post" itemscope itemprop="blogPost">
    
    <div class="article-meta">
      <a href="" class="article-date">
      <time datetime="2021-09-03T09:39:59.000Z" itemprop="datePublished">2021-09-03</time>
</a>
    </div>
  
  <div class="article-inner">
    
      <input type="hidden" class="isFancy" />
    
    
      <header class="article-header">
        
  
    <h1 class="article-title" itemprop="name">
      元数据平台-数据血缘
    </h1>
  

      </header>
      
      <div class="article-info article-info-post">
        
    <div class="article-category tagcloud">
    <a class="article-category-link" href="categories/元数据平台/">元数据平台</a>
    </div>


        
    <div class="article-tag tagcloud">
        <ul class="article-tag-list"><li class="article-tag-list-item"><a class="article-tag-list-link" href="tags/SQL解析/">SQL解析</a></li><li class="article-tag-list-item"><a class="article-tag-list-link" href="tags/SQL采集/">SQL采集</a></li><li class="article-tag-list-item"><a class="article-tag-list-link" href="tags/hive/">hive</a></li><li class="article-tag-list-item"><a class="article-tag-list-link" href="tags/大数据/">大数据</a></li><li class="article-tag-list-item"><a class="article-tag-list-link" href="tags/数据血缘/">数据血缘</a></li></ul>
    </div>

        <div class="clearfix"></div>
      </div>
      
    
    <div class="article-entry" itemprop="articleBody">
      
          
        <h3 id="架构"><a href="#架构" class="headerlink" title="架构"></a>架构</h3><p><img src="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210903154320404.png" alt="image-20210903154320404"></p>
<p>首先，我们先了解一下元数据平台架构，主流程是：SQL采集 –》 SQL解析 –》应用层。</p>
<ol>
<li>SQL采集：针对各种SQL查询引擎，编写相应的钩子函数进行SQL收集，收集内容有执行时间、执行耗时、执行用户、执行引擎、jobId和执行SQL等等，最后把SQL信息入Kafka。Hive是实现ExecuteWithHookContext接口，Presto是实现EventListener接口，Spark是实现SparkListner接口。</li>
<li>SQL解析：Flink实时消费Kafka数据，进行SQL解析。解析SQL的过程为：定义词法规则和语法规则文件 –》 使用Antlr实现SQL词法和语法解析 –》生成AST语法树 –》遍历AST语法树，考虑到Presto和Spark的SQL语法类似，因此直接参考Hive底层源码实现SQL解析。解析完成后，把血缘信息和元数据信息分别入JanusGraph和ElasticSearch。</li>
</ol>
<a id="more"></a>
<ol start="3">
<li>应用层：数据录入JanusGraph和ElasticSearch后，就可以进行血缘查询和元数据查询。然后通过graph节点的出度入度进行热点分析。</li>
</ol>
<h3 id="SQL采集"><a href="#SQL采集" class="headerlink" title="SQL采集"></a>SQL采集</h3><p>本文以Hive的SQL采集为例，编写自定义Hive Hook类，实现SQL采集。此Hook类可通过hive-site.xml配置实现SQL信息入Kafka或JDBC存储引擎。</p>
<h4 id="SqlCollectionHook"><a href="#SqlCollectionHook" class="headerlink" title="SqlCollectionHook"></a>SqlCollectionHook</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.hive;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.google.common.base.Splitter;</span><br><span class="line"><span class="keyword">import</span> com.google.common.base.Strings;</span><br><span class="line"><span class="keyword">import</span> com.google.common.collect.Sets;</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity;</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.strategy.JdbcSqlCollectionStrategy;</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.strategy.KafkaSqlCollectionStrategy;</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.strategy.SqlCollectionStrategy;</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.lang3.time.DateFormatUtils;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.conf.HiveConf;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.QueryPlan;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.exec.TaskRunner;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.hooks.HookContext;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.session.SessionState;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util.Date;</span><br><span class="line"><span class="keyword">import</span> java.util.List;</span><br><span class="line"><span class="keyword">import</span> java.util.Set;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@date</span> 2021-08-16 10:22</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SqlCollectionHook</span> <span class="keyword">implements</span> <span class="title">ExecuteWithHookContext</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String DATE_TIME_FORMAT = <span class="string">"yyyy-MM-dd HH:mm:ss"</span>;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_SQL_STRATEGY = <span class="string">"lineage.sql.strategy"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_SQL_STRATEGY_JDBC = <span class="string">"jdbc"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_SQL_STRATEGY_KAFKA = <span class="string">"kafka"</span>;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">(HookContext hookContext)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">assert</span> (hookContext.getHookType() == HookContext.HookType.POST_EXEC_HOOK);</span><br><span class="line">        QueryPlan plan = hookContext.getQueryPlan();</span><br><span class="line">        SessionState ss = SessionState.get();</span><br><span class="line"></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line"></span><br><span class="line">            HiveConf conf = ss.getConf();</span><br><span class="line">            <span class="keyword">boolean</span> testMode = conf.getBoolVar(HiveConf.ConfVars.HIVE_IN_TEST);</span><br><span class="line"></span><br><span class="line">            <span class="comment">// 获取hook类名</span></span><br><span class="line">            Set&lt;String&gt; postExecHooks = Sets.newHashSet(Splitter.on(<span class="string">","</span>).trimResults()</span><br><span class="line">                    .omitEmptyStrings()</span><br><span class="line">                    .split(Strings.nullToEmpty(HiveConf.getVar(conf, HiveConf.ConfVars.POSTEXECHOOKS))));</span><br><span class="line">            <span class="keyword">if</span> (!testMode &amp;&amp; postExecHooks.contains(<span class="keyword">this</span>.getClass().getName())) &#123;</span><br><span class="line">                <span class="comment">// Don't emit user/timestamp info in test mode,</span></span><br><span class="line">                <span class="comment">// so that the test golden output file is fixed.</span></span><br><span class="line"></span><br><span class="line">                <span class="keyword">long</span> queryTime = plan.getQueryStartTime().longValue();</span><br><span class="line">                <span class="keyword">if</span> (queryTime == <span class="number">0</span>) &#123;</span><br><span class="line">                    queryTime = System.currentTimeMillis();</span><br><span class="line">                &#125;</span><br><span class="line"></span><br><span class="line">                String logtime = DateFormatUtils.format(<span class="keyword">new</span> Date(queryTime), DATE_TIME_FORMAT);</span><br><span class="line">                String sql = plan.getQueryStr();</span><br><span class="line">                String user = hookContext.getUgi().getUserName();</span><br><span class="line">                <span class="keyword">long</span> timestamp = queryTime;</span><br><span class="line">                <span class="keyword">long</span> duration = System.currentTimeMillis() - queryTime;</span><br><span class="line"></span><br><span class="line">                StringBuilder jobIdStr = <span class="keyword">new</span> StringBuilder();</span><br><span class="line">                List&lt;TaskRunner&gt; tasks = hookContext.getCompleteTaskList();</span><br><span class="line">                <span class="keyword">if</span> (tasks != <span class="keyword">null</span> &amp;&amp; !tasks.isEmpty()) &#123;</span><br><span class="line">                    <span class="keyword">for</span> (TaskRunner task : tasks) &#123;</span><br><span class="line">                        String jobId = task.getTask().getJobID();</span><br><span class="line">                        <span class="keyword">if</span> (jobId != <span class="keyword">null</span>) &#123;</span><br><span class="line">                            jobIdStr.append(jobId).append(<span class="string">","</span>);</span><br><span class="line">                        &#125;</span><br><span class="line">                    &#125;</span><br><span class="line">                    <span class="keyword">if</span> (jobIdStr.length() &gt; <span class="number">0</span>) &#123;</span><br><span class="line">                        jobIdStr.deleteCharAt(jobIdStr.length() - <span class="number">1</span>);</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">                String jobIds = jobIdStr.toString();</span><br><span class="line">                String engine = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_EXECUTION_ENGINE);</span><br><span class="line">                String database = ss.getCurrentDatabase();</span><br><span class="line">                String datasource = <span class="string">"hive"</span>;</span><br><span class="line"></span><br><span class="line">                SqlEntity sqlEntity = SqlEntity.newBuilder()</span><br><span class="line">                        .logtime(logtime)</span><br><span class="line">                        .sql(sql)</span><br><span class="line">                        .user(user)</span><br><span class="line">                        .timestamp(timestamp)</span><br><span class="line">                        .duration(duration)</span><br><span class="line">                        .jobids(jobIds)</span><br><span class="line">                        .engine(engine)</span><br><span class="line">                        .database(database)</span><br><span class="line">                        .datasource(datasource)</span><br><span class="line">                        .build();</span><br><span class="line"></span><br><span class="line">                SqlCollectionStrategy strategy = getStrategy(conf);</span><br><span class="line">                <span class="keyword">if</span> (strategy != <span class="keyword">null</span>) &#123;</span><br><span class="line">                    strategy.invoke(conf, sqlEntity);</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">        &#125; <span class="keyword">catch</span> (Throwable t) &#123;</span><br><span class="line">            <span class="comment">// Don't fail the query just because of any lineage issue.</span></span><br><span class="line">            log(<span class="string">"Failed to save lineage sql\n"</span></span><br><span class="line">                    + org.apache.hadoop.util.StringUtils.stringifyException(t));</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Logger an error to console if available.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">log</span><span class="params">(String error)</span> </span>&#123;</span><br><span class="line">        SessionState.LogHelper console = SessionState.getConsole();</span><br><span class="line">        <span class="keyword">if</span> (console != <span class="keyword">null</span>) &#123;</span><br><span class="line">            console.printError(error);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> SqlCollectionStrategy <span class="title">getStrategy</span><span class="params">(HiveConf conf)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">switch</span> (conf.get(LINEAGE_SQL_STRATEGY)) &#123;</span><br><span class="line">            <span class="keyword">case</span> LINEAGE_SQL_STRATEGY_JDBC:</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">new</span> JdbcSqlCollectionStrategy();</span><br><span class="line">            <span class="keyword">case</span> LINEAGE_SQL_STRATEGY_KAFKA:</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">new</span> KafkaSqlCollectionStrategy();</span><br><span class="line">            <span class="keyword">default</span>:</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">null</span>;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="SqlCollectionStrategy"><a href="#SqlCollectionStrategy" class="headerlink" title="SqlCollectionStrategy"></a>SqlCollectionStrategy</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.hive.strategy;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.conf.HiveConf;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@date</span> 2021-08-17 10:27</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">interface</span> <span class="title">SqlCollectionStrategy</span> </span>&#123;</span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * 执行策略</span></span><br><span class="line"><span class="comment">     *</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> conf HiveConf</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@param</span> sqlEntity sql</span></span><br><span class="line"><span class="comment">     * <span class="doctag">@return</span> bool</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">boolean</span> <span class="title">invoke</span><span class="params">(HiveConf conf, SqlEntity sqlEntity)</span></span>;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="KafkaSqlCollectionStrategy"><a href="#KafkaSqlCollectionStrategy" class="headerlink" title="KafkaSqlCollectionStrategy"></a>KafkaSqlCollectionStrategy</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.hive.strategy;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.conf.HiveConf;</span><br><span class="line"><span class="keyword">import</span> org.apache.kafka.clients.producer.KafkaProducer;</span><br><span class="line"><span class="keyword">import</span> org.apache.kafka.clients.producer.ProducerRecord;</span><br><span class="line"><span class="keyword">import</span> org.codehaus.jackson.map.ObjectMapper;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.Logger;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.LoggerFactory;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util.Properties;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@date</span> 2021-08-17 10:52</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">KafkaSqlCollectionStrategy</span> <span class="keyword">implements</span> <span class="title">SqlCollectionStrategy</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> Logger LOG = LoggerFactory.getLogger(KafkaSqlCollectionStrategy.class);</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_BOOTSTRAP_SERVERS = <span class="string">"lineage.bootstrap.servers"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_TOPIC = <span class="string">"lineage.topic"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> ObjectMapper MAPPER = <span class="keyword">new</span> ObjectMapper();</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">invoke</span><span class="params">(HiveConf conf, SqlEntity sqlEntity)</span> </span>&#123;</span><br><span class="line">        Properties properties = <span class="keyword">new</span> Properties();</span><br><span class="line">        properties.put(<span class="string">"bootstrap.servers"</span>, conf.get(LINEAGE_BOOTSTRAP_SERVERS));</span><br><span class="line">        properties.put(<span class="string">"key.serializer"</span>, <span class="string">"org.apache.kafka.common.serialization.StringSerializer"</span>);</span><br><span class="line">        properties.put(<span class="string">"value.serializer"</span>, <span class="string">"org.apache.kafka.common.serialization.StringSerializer"</span>);</span><br><span class="line">        <span class="keyword">try</span> (KafkaProducer&lt;String, String&gt; producer = <span class="keyword">new</span> KafkaProducer&lt;&gt;(properties)) &#123;</span><br><span class="line">            String topic = conf.get(LINEAGE_TOPIC);</span><br><span class="line">            producer.send(<span class="keyword">new</span> ProducerRecord&lt;&gt;(topic, MAPPER.writeValueAsString(sqlEntity)));</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">true</span>;</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            LOG.error(<span class="string">"Error send linage sql to kafka, "</span> + e.getMessage());</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="JdbcSqlCollectionStrategy"><a href="#JdbcSqlCollectionStrategy" class="headerlink" title="JdbcSqlCollectionStrategy"></a>JdbcSqlCollectionStrategy</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.hive.strategy;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.conf.HiveConf;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.Logger;</span><br><span class="line"><span class="keyword">import</span> org.slf4j.LoggerFactory;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.sql.*;</span><br><span class="line"><span class="keyword">import</span> java.util.Properties;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@date</span> 2021-08-17 10:34</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JdbcSqlCollectionStrategy</span> <span class="keyword">implements</span> <span class="title">SqlCollectionStrategy</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> Logger LOG = LoggerFactory.getLogger(JdbcSqlCollectionStrategy.class);</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_DB_DRIVER = <span class="string">"lineage.db.driver"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_DB_URL = <span class="string">"lineage.db.url"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_DB_USER = <span class="string">"lineage.db.user"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_DB_PASSWORD = <span class="string">"lineage.db.password"</span>;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> <span class="keyword">final</span> String LINEAGE_DB_TABLE = <span class="string">"lineage.db.table"</span>;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">invoke</span><span class="params">(HiveConf conf, SqlEntity sqlEntity)</span> </span>&#123;</span><br><span class="line">        Connection conn = <span class="keyword">null</span>;</span><br><span class="line">        PreparedStatement pstmt = <span class="keyword">null</span>;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            conn = getConnection(conf);</span><br><span class="line">            pstmt = conn.prepareStatement(getInsertLineageSql(conf));</span><br><span class="line"></span><br><span class="line">            pstmt.setString(<span class="number">1</span>, sqlEntity.getLogtime());</span><br><span class="line">            pstmt.setString(<span class="number">2</span>, sqlEntity.getSql());</span><br><span class="line">            pstmt.setString(<span class="number">3</span>, sqlEntity.getUser());</span><br><span class="line">            pstmt.setLong(<span class="number">4</span>, sqlEntity.getTimestamp());</span><br><span class="line">            pstmt.setLong(<span class="number">5</span>, sqlEntity.getDuration());</span><br><span class="line">            pstmt.setString(<span class="number">6</span>, sqlEntity.getJobids());</span><br><span class="line">            pstmt.setString(<span class="number">7</span>, sqlEntity.getEngine());</span><br><span class="line">            pstmt.setString(<span class="number">8</span>, sqlEntity.getDatabase());</span><br><span class="line">            pstmt.setString(<span class="number">9</span>, sqlEntity.getDatasource());</span><br><span class="line"></span><br><span class="line">            <span class="keyword">return</span> pstmt.execute();</span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            LOG.error(<span class="string">"Error insert lineage sql table, "</span> + e.getMessage());</span><br><span class="line">        &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">            closeResources(conn, pstmt, <span class="keyword">null</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> Connection <span class="title">getConnection</span><span class="params">(HiveConf conf)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">        String jdbcDriver = conf.get(LINEAGE_DB_DRIVER);</span><br><span class="line">        Driver driver = (Driver) Class.forName(jdbcDriver).newInstance();</span><br><span class="line">        Properties prop = <span class="keyword">new</span> Properties();</span><br><span class="line">        prop.setProperty(<span class="string">"user"</span>, conf.get(LINEAGE_DB_USER));</span><br><span class="line">        prop.setProperty(<span class="string">"password"</span>, conf.get(LINEAGE_DB_PASSWORD));</span><br><span class="line">        Connection conn = driver.connect(conf.get(LINEAGE_DB_URL), prop);</span><br><span class="line">        conn.setAutoCommit(<span class="keyword">true</span>);</span><br><span class="line">        <span class="keyword">return</span> conn;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> String <span class="title">getInsertLineageSql</span><span class="params">(HiveConf conf)</span> </span>&#123;</span><br><span class="line">        String table = conf.get(LINEAGE_DB_TABLE);</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> <span class="string">"insert into "</span> + table</span><br><span class="line">                + <span class="string">"(`logtime`, `sql`, `user`, `timestamp`, `duration`, `jobids`, `engine`, `database`, `datasource`) "</span></span><br><span class="line">                + <span class="string">"values(?, ?, ?, ?, ?, ?, ?, ?, ?)"</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">closeResources</span><span class="params">(Connection conn, Statement stmt, ResultSet rs)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span> (rs != <span class="keyword">null</span>) &#123;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                rs.close();</span><br><span class="line">            &#125; <span class="keyword">catch</span> (SQLException e) &#123;</span><br><span class="line">                LOG.error(<span class="string">"Error closing ResultSet: "</span> + e.getMessage());</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">if</span> (stmt != <span class="keyword">null</span>) &#123;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                stmt.close();</span><br><span class="line">            &#125; <span class="keyword">catch</span> (SQLException e) &#123;</span><br><span class="line">                System.err.println(<span class="string">"Error closing Statement: "</span> + e.getMessage());</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">if</span> (conn != <span class="keyword">null</span>) &#123;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                conn.close();</span><br><span class="line">            &#125; <span class="keyword">catch</span> (SQLException e) &#123;</span><br><span class="line">                System.err.println(<span class="string">"Error closing Connection: "</span> + e.getMessage());</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="hive-site-xml"><a href="#hive-site-xml" class="headerlink" title="hive-site.xml"></a>hive-site.xml</h4><p>添加自定义Hook类配置，有两种SQL存储策略：Kafka和JDBC，选择一种配置。</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">configuration</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;<span class="name">property</span>&gt;</span></span><br><span class="line"> 		 <span class="tag">&lt;<span class="name">name</span>&gt;</span>hive.exec.post.hooks<span class="tag">&lt;/<span class="name">name</span>&gt;</span></span><br><span class="line"> 		 <span class="tag">&lt;<span class="name">value</span>&gt;</span>com.xyueji.traffic.hive.SqlCollectionHook<span class="tag">&lt;/<span class="name">value</span>&gt;</span></span><br><span class="line">  	 <span class="tag">&lt;<span class="name">description</span>&gt;</span>Comma-separated list of post-execution hooks to be invoked for each statement.  A post-execution hook is specified as the name of a Java class which implements the org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext interface.<span class="tag">&lt;/<span class="name">description</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line">  <span class="comment">&lt;!--kafka存储策略--&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span>&gt;</span></span><br><span class="line">  		<span class="tag">&lt;<span class="name">name</span>&gt;</span>lineage.sql.strategy<span class="tag">&lt;/<span class="name">name</span>&gt;</span></span><br><span class="line">  		<span class="tag">&lt;<span class="name">value</span>&gt;</span>kafka<span class="tag">&lt;/<span class="name">value</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span>&gt;</span></span><br><span class="line">  		<span class="tag">&lt;<span class="name">name</span>&gt;</span>lineage.bootstrap.servers<span class="tag">&lt;/<span class="name">name</span>&gt;</span></span><br><span class="line">  		<span class="tag">&lt;<span class="name">value</span>&gt;</span>localhost:9092<span class="tag">&lt;/<span class="name">value</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span>&gt;</span></span><br><span class="line">  		<span class="tag">&lt;<span class="name">name</span>&gt;</span>lineage.topic<span class="tag">&lt;/<span class="name">name</span>&gt;</span></span><br><span class="line">  		<span class="tag">&lt;<span class="name">value</span>&gt;</span>lineage_sql<span class="tag">&lt;/<span class="name">value</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line">  <span class="comment">&lt;!--jdbc存储策略--&gt;</span></span><br><span class="line">  <span class="comment">&lt;!--</span></span><br><span class="line"><span class="comment">  &lt;property&gt;</span></span><br><span class="line"><span class="comment">  		&lt;name&gt;lineage.sql.strategy&lt;/name&gt;</span></span><br><span class="line"><span class="comment">  		&lt;value&gt;jdbc&lt;/value&gt;</span></span><br><span class="line"><span class="comment">	&lt;/property&gt;</span></span><br><span class="line"><span class="comment">	&lt;property&gt;</span></span><br><span class="line"><span class="comment">  		&lt;name&gt;lineage.db.driver&lt;/name&gt;</span></span><br><span class="line"><span class="comment">  		&lt;value&gt;com.mysql.jdbc.Driver&lt;/value&gt;</span></span><br><span class="line"><span class="comment">	&lt;/property&gt;</span></span><br><span class="line"><span class="comment">	&lt;property&gt;</span></span><br><span class="line"><span class="comment">  		&lt;name&gt;lineage.db.url&lt;/name&gt;</span></span><br><span class="line"><span class="comment">  		&lt;value&gt;jdbc:mysql://<span class="doctag">xxx:</span>3306/traffic?useUnicode=true&amp;characterEncoding=utf-8&lt;/value&gt;</span></span><br><span class="line"><span class="comment">	&lt;/property&gt;</span></span><br><span class="line"><span class="comment">	&lt;property&gt;</span></span><br><span class="line"><span class="comment">  		&lt;name&gt;lineage.db.user&lt;/name&gt;</span></span><br><span class="line"><span class="comment">  		&lt;value&gt;xxx&lt;/value&gt;</span></span><br><span class="line"><span class="comment">	&lt;/property&gt;</span></span><br><span class="line"><span class="comment">	&lt;property&gt;</span></span><br><span class="line"><span class="comment">  		&lt;name&gt;lineage.db.password&lt;/name&gt;</span></span><br><span class="line"><span class="comment">  		&lt;value&gt;xxx&lt;/value&gt;</span></span><br><span class="line"><span class="comment">	&lt;/property&gt;</span></span><br><span class="line"><span class="comment">	&lt;property&gt;</span></span><br><span class="line"><span class="comment">  		&lt;name&gt;lineage.db.table&lt;/name&gt;</span></span><br><span class="line"><span class="comment">  		&lt;value&gt;lineage_sql&lt;/value&gt;</span></span><br><span class="line"><span class="comment">	&lt;/property&gt;</span></span><br><span class="line"><span class="comment">	--&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">configuration</span>&gt;</span></span><br></pre></td></tr></table></figure>
<blockquote>
<p>重启Hive metastore使Hive Hook生效</p>
</blockquote>
<h3 id="SQL解析"><a href="#SQL解析" class="headerlink" title="SQL解析"></a>SQL解析</h3><p>参考Hive源码模仿编写SQL解析工具，先经过类进行语法分析，再根据Schema生成执行计划QueryPlan。字段血缘可根据LineageLogger类进行实现，表血缘根据字段血缘获取。</p>
<h4 id="SQLParse2"><a href="#SQLParse2" class="headerlink" title="SQLParse2"></a>SQLParse2</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br><span class="line">157</span><br><span class="line">158</span><br><span class="line">159</span><br><span class="line">160</span><br><span class="line">161</span><br><span class="line">162</span><br><span class="line">163</span><br><span class="line">164</span><br><span class="line">165</span><br><span class="line">166</span><br><span class="line">167</span><br><span class="line">168</span><br><span class="line">169</span><br><span class="line">170</span><br><span class="line">171</span><br><span class="line">172</span><br><span class="line">173</span><br><span class="line">174</span><br><span class="line">175</span><br><span class="line">176</span><br><span class="line">177</span><br><span class="line">178</span><br><span class="line">179</span><br><span class="line">180</span><br><span class="line">181</span><br><span class="line">182</span><br><span class="line">183</span><br><span class="line">184</span><br><span class="line">185</span><br><span class="line">186</span><br><span class="line">187</span><br><span class="line">188</span><br><span class="line">189</span><br><span class="line">190</span><br><span class="line">191</span><br><span class="line">192</span><br><span class="line">193</span><br><span class="line">194</span><br><span class="line">195</span><br><span class="line">196</span><br><span class="line">197</span><br><span class="line">198</span><br><span class="line">199</span><br><span class="line">200</span><br><span class="line">201</span><br><span class="line">202</span><br><span class="line">203</span><br><span class="line">204</span><br><span class="line">205</span><br><span class="line">206</span><br><span class="line">207</span><br><span class="line">208</span><br><span class="line">209</span><br><span class="line">210</span><br><span class="line">211</span><br><span class="line">212</span><br><span class="line">213</span><br><span class="line">214</span><br><span class="line">215</span><br><span class="line">216</span><br><span class="line">217</span><br><span class="line">218</span><br><span class="line">219</span><br><span class="line">220</span><br><span class="line">221</span><br><span class="line">222</span><br><span class="line">223</span><br><span class="line">224</span><br><span class="line">225</span><br><span class="line">226</span><br><span class="line">227</span><br><span class="line">228</span><br><span class="line">229</span><br><span class="line">230</span><br><span class="line">231</span><br><span class="line">232</span><br><span class="line">233</span><br><span class="line">234</span><br><span class="line">235</span><br><span class="line">236</span><br><span class="line">237</span><br><span class="line">238</span><br><span class="line">239</span><br><span class="line">240</span><br><span class="line">241</span><br><span class="line">242</span><br><span class="line">243</span><br><span class="line">244</span><br><span class="line">245</span><br><span class="line">246</span><br><span class="line">247</span><br><span class="line">248</span><br><span class="line">249</span><br><span class="line">250</span><br><span class="line">251</span><br><span class="line">252</span><br><span class="line">253</span><br><span class="line">254</span><br><span class="line">255</span><br><span class="line">256</span><br><span class="line">257</span><br><span class="line">258</span><br><span class="line">259</span><br><span class="line">260</span><br><span class="line">261</span><br><span class="line">262</span><br><span class="line">263</span><br><span class="line">264</span><br><span class="line">265</span><br><span class="line">266</span><br><span class="line">267</span><br><span class="line">268</span><br><span class="line">269</span><br><span class="line">270</span><br><span class="line">271</span><br><span class="line">272</span><br><span class="line">273</span><br><span class="line">274</span><br><span class="line">275</span><br><span class="line">276</span><br><span class="line">277</span><br><span class="line">278</span><br><span class="line">279</span><br><span class="line">280</span><br><span class="line">281</span><br><span class="line">282</span><br><span class="line">283</span><br><span class="line">284</span><br><span class="line">285</span><br><span class="line">286</span><br><span class="line">287</span><br><span class="line">288</span><br><span class="line">289</span><br><span class="line">290</span><br><span class="line">291</span><br><span class="line">292</span><br><span class="line">293</span><br><span class="line">294</span><br><span class="line">295</span><br><span class="line">296</span><br><span class="line">297</span><br><span class="line">298</span><br><span class="line">299</span><br><span class="line">300</span><br><span class="line">301</span><br><span class="line">302</span><br><span class="line">303</span><br><span class="line">304</span><br><span class="line">305</span><br><span class="line">306</span><br><span class="line">307</span><br><span class="line">308</span><br><span class="line">309</span><br><span class="line">310</span><br><span class="line">311</span><br><span class="line">312</span><br><span class="line">313</span><br><span class="line">314</span><br><span class="line">315</span><br><span class="line">316</span><br><span class="line">317</span><br><span class="line">318</span><br><span class="line">319</span><br><span class="line">320</span><br><span class="line">321</span><br><span class="line">322</span><br><span class="line">323</span><br><span class="line">324</span><br><span class="line">325</span><br><span class="line">326</span><br><span class="line">327</span><br><span class="line">328</span><br><span class="line">329</span><br><span class="line">330</span><br><span class="line">331</span><br><span class="line">332</span><br><span class="line">333</span><br><span class="line">334</span><br><span class="line">335</span><br><span class="line">336</span><br><span class="line">337</span><br><span class="line">338</span><br><span class="line">339</span><br><span class="line">340</span><br><span class="line">341</span><br><span class="line">342</span><br><span class="line">343</span><br><span class="line">344</span><br><span class="line">345</span><br><span class="line">346</span><br><span class="line">347</span><br><span class="line">348</span><br><span class="line">349</span><br><span class="line">350</span><br><span class="line">351</span><br><span class="line">352</span><br><span class="line">353</span><br><span class="line">354</span><br><span class="line">355</span><br><span class="line">356</span><br><span class="line">357</span><br><span class="line">358</span><br><span class="line">359</span><br><span class="line">360</span><br><span class="line">361</span><br><span class="line">362</span><br><span class="line">363</span><br><span class="line">364</span><br><span class="line">365</span><br><span class="line">366</span><br><span class="line">367</span><br><span class="line">368</span><br><span class="line">369</span><br><span class="line">370</span><br><span class="line">371</span><br><span class="line">372</span><br><span class="line">373</span><br><span class="line">374</span><br><span class="line">375</span><br><span class="line">376</span><br><span class="line">377</span><br><span class="line">378</span><br><span class="line">379</span><br><span class="line">380</span><br><span class="line">381</span><br><span class="line">382</span><br><span class="line">383</span><br><span class="line">384</span><br><span class="line">385</span><br><span class="line">386</span><br><span class="line">387</span><br><span class="line">388</span><br><span class="line">389</span><br><span class="line">390</span><br><span class="line">391</span><br><span class="line">392</span><br><span class="line">393</span><br><span class="line">394</span><br><span class="line">395</span><br><span class="line">396</span><br><span class="line">397</span><br><span class="line">398</span><br><span class="line">399</span><br><span class="line">400</span><br><span class="line">401</span><br><span class="line">402</span><br><span class="line">403</span><br><span class="line">404</span><br><span class="line">405</span><br><span class="line">406</span><br><span class="line">407</span><br><span class="line">408</span><br><span class="line">409</span><br><span class="line">410</span><br><span class="line">411</span><br><span class="line">412</span><br><span class="line">413</span><br><span class="line">414</span><br><span class="line">415</span><br><span class="line">416</span><br><span class="line">417</span><br><span class="line">418</span><br><span class="line">419</span><br><span class="line">420</span><br><span class="line">421</span><br><span class="line">422</span><br><span class="line">423</span><br><span class="line">424</span><br><span class="line">425</span><br><span class="line">426</span><br><span class="line">427</span><br><span class="line">428</span><br><span class="line">429</span><br><span class="line">430</span><br><span class="line">431</span><br><span class="line">432</span><br><span class="line">433</span><br><span class="line">434</span><br><span class="line">435</span><br><span class="line">436</span><br><span class="line">437</span><br><span class="line">438</span><br><span class="line">439</span><br><span class="line">440</span><br><span class="line">441</span><br><span class="line">442</span><br><span class="line">443</span><br><span class="line">444</span><br><span class="line">445</span><br><span class="line">446</span><br><span class="line">447</span><br><span class="line">448</span><br><span class="line">449</span><br><span class="line">450</span><br><span class="line">451</span><br><span class="line">452</span><br><span class="line">453</span><br><span class="line">454</span><br><span class="line">455</span><br><span class="line">456</span><br><span class="line">457</span><br><span class="line">458</span><br><span class="line">459</span><br><span class="line">460</span><br><span class="line">461</span><br><span class="line">462</span><br><span class="line">463</span><br><span class="line">464</span><br><span class="line">465</span><br><span class="line">466</span><br><span class="line">467</span><br><span class="line">468</span><br><span class="line">469</span><br><span class="line">470</span><br><span class="line">471</span><br><span class="line">472</span><br><span class="line">473</span><br><span class="line">474</span><br><span class="line">475</span><br><span class="line">476</span><br><span class="line">477</span><br><span class="line">478</span><br><span class="line">479</span><br><span class="line">480</span><br><span class="line">481</span><br><span class="line">482</span><br><span class="line">483</span><br><span class="line">484</span><br><span class="line">485</span><br><span class="line">486</span><br><span class="line">487</span><br><span class="line">488</span><br><span class="line">489</span><br><span class="line">490</span><br><span class="line">491</span><br><span class="line">492</span><br><span class="line">493</span><br><span class="line">494</span><br><span class="line">495</span><br><span class="line">496</span><br><span class="line">497</span><br><span class="line">498</span><br><span class="line">499</span><br><span class="line">500</span><br><span class="line">501</span><br><span class="line">502</span><br><span class="line">503</span><br><span class="line">504</span><br><span class="line">505</span><br><span class="line">506</span><br><span class="line">507</span><br><span class="line">508</span><br><span class="line">509</span><br><span class="line">510</span><br><span class="line">511</span><br><span class="line">512</span><br><span class="line">513</span><br><span class="line">514</span><br><span class="line">515</span><br><span class="line">516</span><br><span class="line">517</span><br><span class="line">518</span><br><span class="line">519</span><br><span class="line">520</span><br><span class="line">521</span><br><span class="line">522</span><br><span class="line">523</span><br><span class="line">524</span><br><span class="line">525</span><br><span class="line">526</span><br><span class="line">527</span><br><span class="line">528</span><br><span class="line">529</span><br><span class="line">530</span><br><span class="line">531</span><br><span class="line">532</span><br><span class="line">533</span><br><span class="line">534</span><br><span class="line">535</span><br><span class="line">536</span><br><span class="line">537</span><br><span class="line">538</span><br><span class="line">539</span><br><span class="line">540</span><br><span class="line">541</span><br><span class="line">542</span><br><span class="line">543</span><br><span class="line">544</span><br><span class="line">545</span><br><span class="line">546</span><br><span class="line">547</span><br><span class="line">548</span><br><span class="line">549</span><br><span class="line">550</span><br><span class="line">551</span><br><span class="line">552</span><br><span class="line">553</span><br><span class="line">554</span><br><span class="line">555</span><br><span class="line">556</span><br><span class="line">557</span><br><span class="line">558</span><br><span class="line">559</span><br><span class="line">560</span><br><span class="line">561</span><br><span class="line">562</span><br><span class="line">563</span><br><span class="line">564</span><br><span class="line">565</span><br><span class="line">566</span><br><span class="line">567</span><br><span class="line">568</span><br><span class="line">569</span><br><span class="line">570</span><br><span class="line">571</span><br><span class="line">572</span><br><span class="line">573</span><br><span class="line">574</span><br><span class="line">575</span><br><span class="line">576</span><br><span class="line">577</span><br><span class="line">578</span><br><span class="line">579</span><br><span class="line">580</span><br><span class="line">581</span><br><span class="line">582</span><br><span class="line">583</span><br><span class="line">584</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.hive;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.google.common.collect.Lists;</span><br><span class="line"><span class="keyword">import</span> com.google.gson.stream.JsonWriter;</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.codec.digest.DigestUtils;</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.collections.SetUtils;</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.io.output.StringBuilderWriter;</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.lang3.StringUtils;</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.lang3.time.DateFormatUtils;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.common.ObjectPair;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.conf.HiveConf;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.metastore.api.FieldSchema;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.metastore.api.Schema;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.Context;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.Driver;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.QueryPlan;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.QueryState;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.exec.ColumnInfo;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.exec.SelectOperator;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.exec.Utilities;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.hooks.Entity;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.hooks.LineageInfo;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.hooks.WriteEntity;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.metadata.Table;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.optimizer.lineage.LineageCtx.Index;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.parse.ASTNode;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.parse.ParseUtils;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.parse.SemanticAnalyzerFactory;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.plan.HiveOperation;</span><br><span class="line"><span class="keyword">import</span> org.apache.hadoop.hive.ql.session.SessionState;</span><br><span class="line"><span class="keyword">import</span> org.joor.Reflect;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.io.IOException;</span><br><span class="line"><span class="keyword">import</span> java.io.Serializable;</span><br><span class="line"><span class="keyword">import</span> java.util.*;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@date</span> 2021-08-11 11:58</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SQLParse2</span> <span class="keyword">implements</span> <span class="title">Serializable</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">final</span> <span class="keyword">static</span> String DATE_FORMAT = <span class="string">"yyyy-MM-dd HH:mm:ss"</span>;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">final</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">Edge</span> <span class="keyword">implements</span> <span class="title">Serializable</span> </span>&#123;</span><br><span class="line">        <span class="keyword">private</span> Set&lt;Vertex&gt; sources;</span><br><span class="line">        <span class="keyword">private</span> Set&lt;Vertex&gt; targets;</span><br><span class="line">        <span class="keyword">private</span> String expr;</span><br><span class="line">        <span class="keyword">private</span> HiveOperation operation;</span><br><span class="line"></span><br><span class="line">        Edge(Set&lt;Vertex&gt; sources, Set&lt;Vertex&gt; targets, String expr, HiveOperation operation) &#123;</span><br><span class="line">            <span class="keyword">this</span>.sources = sources;</span><br><span class="line">            <span class="keyword">this</span>.targets = targets;</span><br><span class="line">            <span class="keyword">this</span>.expr = expr;</span><br><span class="line">            <span class="keyword">this</span>.operation = operation;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Set&lt;Vertex&gt; <span class="title">getSources</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> sources;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setSources</span><span class="params">(Set&lt;Vertex&gt; sources)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.sources = sources;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Set&lt;Vertex&gt; <span class="title">getTargets</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> targets;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setTargets</span><span class="params">(Set&lt;Vertex&gt; targets)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.targets = targets;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getExpr</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> expr;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setExpr</span><span class="params">(String expr)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.expr = expr;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> HiveOperation <span class="title">getOperation</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> operation;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setOperation</span><span class="params">(HiveOperation operation)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.operation = operation;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">final</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">Vertex</span> <span class="keyword">implements</span> <span class="title">Serializable</span> </span>&#123;</span><br><span class="line">        <span class="keyword">public</span> <span class="keyword">enum</span> Type &#123;</span><br><span class="line">            COLUMN, TABLE</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">public</span> <span class="keyword">enum</span> Status &#123;</span><br><span class="line">            OFFLINE, ONLINE</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">private</span> Vertex.Type type;</span><br><span class="line">        <span class="keyword">private</span> String createAt;</span><br><span class="line">        <span class="keyword">private</span> String db;</span><br><span class="line">        <span class="keyword">private</span> String owner;</span><br><span class="line">        <span class="keyword">private</span> String table;</span><br><span class="line">        <span class="keyword">private</span> String tableType;</span><br><span class="line">        <span class="keyword">private</span> String col;</span><br><span class="line">        <span class="keyword">private</span> String colType;</span><br><span class="line">        <span class="keyword">private</span> Vertex.Status status;</span><br><span class="line">        <span class="keyword">private</span> String md5;</span><br><span class="line"></span><br><span class="line">        Vertex(Type type, String createAt, String db, String owner, String table, String tableType, Status status, String md5) &#123;</span><br><span class="line">            <span class="keyword">this</span>.type = type;</span><br><span class="line">            <span class="keyword">this</span>.createAt = createAt;</span><br><span class="line">            <span class="keyword">this</span>.db = db;</span><br><span class="line">            <span class="keyword">this</span>.owner = owner;</span><br><span class="line">            <span class="keyword">this</span>.table = table;</span><br><span class="line">            <span class="keyword">this</span>.tableType = tableType;</span><br><span class="line">            <span class="keyword">this</span>.status = status;</span><br><span class="line">            <span class="keyword">this</span>.md5 = md5;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        Vertex(Type type, String createAt, String db, String owner, String table, String tableType, String col, String colType, Status status, String md5) &#123;</span><br><span class="line">            <span class="keyword">this</span>(type, createAt, db, owner, table, tableType, status, md5);</span><br><span class="line">            <span class="keyword">this</span>.col = col;</span><br><span class="line">            <span class="keyword">this</span>.colType = colType;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="meta">@Override</span></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">int</span> <span class="title">hashCode</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">int</span> result = <span class="number">0</span>;</span><br><span class="line"></span><br><span class="line">            result = result * <span class="number">31</span> + type.hashCode();</span><br><span class="line">            result = result * <span class="number">31</span> + db.hashCode();</span><br><span class="line">            result = result * <span class="number">31</span> + table.hashCode();</span><br><span class="line">            result = result * <span class="number">31</span> + (col == <span class="keyword">null</span> ? <span class="number">0</span> : col.hashCode());</span><br><span class="line">            <span class="keyword">return</span> result;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="meta">@Override</span></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">equals</span><span class="params">(Object obj)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">if</span> (<span class="keyword">this</span> == obj) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">true</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            <span class="keyword">if</span> (!(obj <span class="keyword">instanceof</span> SQLParse2.Vertex)) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">            &#125;</span><br><span class="line">            SQLParse2.Vertex vertex = (SQLParse2.Vertex) obj;</span><br><span class="line"></span><br><span class="line">            <span class="keyword">if</span> (type != <span class="keyword">null</span> &amp;&amp; !type.equals(vertex.type)) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="keyword">if</span> (db != <span class="keyword">null</span> &amp;&amp; !db.equals(vertex.db)) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="keyword">if</span> (table != <span class="keyword">null</span> &amp;&amp; !table.equals(vertex.table)) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="keyword">if</span> (col != <span class="keyword">null</span> &amp;&amp; !col.equals(vertex.col)) &#123;</span><br><span class="line">                <span class="keyword">return</span> <span class="keyword">false</span>;</span><br><span class="line">            &#125;</span><br><span class="line"></span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">true</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Type <span class="title">getType</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> type;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setType</span><span class="params">(Type type)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.type = type;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getCreateAt</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> createAt;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setCreateAt</span><span class="params">(String createAt)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.createAt = createAt;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getDb</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> db;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setDb</span><span class="params">(String db)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.db = db;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getOwner</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> owner;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setOwner</span><span class="params">(String owner)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.owner = owner;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getTable</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> table;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setTable</span><span class="params">(String table)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.table = table;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getTableType</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> tableType;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setTableType</span><span class="params">(String tableType)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.tableType = tableType;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getCol</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> col;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setCol</span><span class="params">(String col)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.col = col;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getColType</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> colType;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setColType</span><span class="params">(String colType)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.colType = colType;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Status <span class="title">getStatus</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> status;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setStatus</span><span class="params">(Status status)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.status = status;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> String <span class="title">getMd5</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> md5;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setMd5</span><span class="params">(String md5)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.md5 = md5;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">static</span> HiveConf conf = <span class="keyword">new</span> HiveConf();</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">init</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.HIVE_AUTHORIZATION_MANAGER,</span><br><span class="line">                <span class="string">"org.apache.hadoop.hive.ql.security.authorization.plugin.sqlstd.SQLStdHiveAuthorizerFactory"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, <span class="string">"nonstrict"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.HIVE_TXN_MANAGER, <span class="string">"org.apache.hadoop.hive.ql.lockmgr.DbTxnManager"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.METASTOREURIS, <span class="string">"thrift://hadoop3:9083"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.DOWNLOADED_RESOURCES_DIR, <span class="string">"/tmp/hive"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.METASTOREWAREHOUSE, <span class="string">"/tmp/user/hive/warehouse"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.LOCALSCRATCHDIR, <span class="string">"/tmp"</span>);</span><br><span class="line">        conf.setVar(HiveConf.ConfVars.POSTEXECHOOKS, <span class="string">"org.apache.hadoop.hive.ql.hooks.PostExecutePrinter"</span>);</span><br><span class="line"></span><br><span class="line">        SessionState.start(conf);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> </span>&#123;</span><br><span class="line">        String sql = <span class="string">"insert overwrite table temp select coalesce(blockid, EXPONENT) blockid_exponent from ods.road_real_time group by blockid, EXPONENT limit 10"</span>;</span><br><span class="line">        SQLParse2 sqlParse2 = <span class="keyword">new</span> SQLParse2();</span><br><span class="line">        sqlParse2.run(sql);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> List&lt;Edge&gt; <span class="title">run</span><span class="params">(String query)</span> </span>&#123;</span><br><span class="line">        init();</span><br><span class="line">        List&lt;Edge&gt; edges = <span class="keyword">new</span> ArrayList&lt;&gt;();</span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">            QueryState queryState = <span class="keyword">new</span> QueryState(conf);</span><br><span class="line">            Context context = <span class="keyword">new</span> Context(conf);</span><br><span class="line"></span><br><span class="line">            context.setHDFSCleanup(<span class="keyword">true</span>);</span><br><span class="line"></span><br><span class="line">            ASTNode astNode = ParseUtils.parse(query, context);</span><br><span class="line"></span><br><span class="line">            BaseSemanticAnalyzer sem = SemanticAnalyzerFactory.get(queryState, astNode);</span><br><span class="line"></span><br><span class="line">            sem.analyze(astNode, context);</span><br><span class="line"></span><br><span class="line">            Schema schema = Reflect.on(Driver.class).call(<span class="string">"getSchema"</span>, sem, conf).get();</span><br><span class="line">            QueryPlan plan = <span class="keyword">new</span> QueryPlan(query, sem, <span class="number">0L</span>, <span class="keyword">null</span>, queryState.getHiveOperation(), schema);</span><br><span class="line"></span><br><span class="line">            Index index = SessionState.get() != <span class="keyword">null</span> ?</span><br><span class="line">                    SessionState.get().getLineageState().getIndex() : <span class="keyword">new</span> Index();</span><br><span class="line"></span><br><span class="line">            edges = getEdges(plan, index);</span><br><span class="line">            Set&lt;Vertex&gt; vertices = getVertices(edges);</span><br><span class="line"></span><br><span class="line">            StringBuilderWriter out = <span class="keyword">new</span> StringBuilderWriter(<span class="number">1024</span>);</span><br><span class="line">            JsonWriter writer = <span class="keyword">new</span> JsonWriter(out);</span><br><span class="line">            writer.beginObject();</span><br><span class="line">            writeEdges(writer, edges);</span><br><span class="line">            writeVertices(writer, vertices);</span><br><span class="line">            writer.endObject();</span><br><span class="line">            writer.close();</span><br><span class="line">            System.out.println(out.toString());</span><br><span class="line"></span><br><span class="line">            <span class="comment">// 此处还可构建任务血缘，根据hive（ExecuteWithHookContext）、spark(SparkListener)、presto(EventListener)收集的sql以及任务信息（可参考LineageLogger）</span></span><br><span class="line">            <span class="comment">// TODO 构建任务血缘</span></span><br><span class="line"></span><br><span class="line">            <span class="comment">// TODO 根据DDL操作更新图信息</span></span><br><span class="line"></span><br><span class="line">        &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">            log(<span class="string">"SQL parse failed, msg is : "</span></span><br><span class="line">                    + org.apache.hadoop.util.StringUtils.stringifyException(e));</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> edges;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Logger an error to console if available.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">log</span><span class="params">(String error)</span> </span>&#123;</span><br><span class="line">        SessionState.LogHelper console = SessionState.getConsole();</span><br><span class="line">        <span class="keyword">if</span> (console != <span class="keyword">null</span>) &#123;</span><br><span class="line">            console.printError(error);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Based on the final select operator, find out all the target columns.</span></span><br><span class="line"><span class="comment">     * For each target column, find out its sources based on the dependency index.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> List&lt;Edge&gt; <span class="title">getEdges</span><span class="params">(QueryPlan plan, Index index)</span> </span>&#123;</span><br><span class="line">        LinkedHashMap&lt;String, ObjectPair&lt;SelectOperator,</span><br><span class="line">                Table&gt;&gt; finalSelOps = index.getFinalSelectOps();</span><br><span class="line">        Map&lt;String, Vertex&gt; vertexCache = <span class="keyword">new</span> LinkedHashMap&lt;String, Vertex&gt;();</span><br><span class="line">        List&lt;Edge&gt; edges = <span class="keyword">new</span> ArrayList&lt;Edge&gt;();</span><br><span class="line">        <span class="keyword">for</span> (ObjectPair&lt;SelectOperator,</span><br><span class="line">                org.apache.hadoop.hive.ql.metadata.Table&gt; pair : finalSelOps.values()) &#123;</span><br><span class="line">            List&lt;FieldSchema&gt; fieldSchemas = <span class="keyword">new</span> ArrayList&lt;&gt;();</span><br><span class="line">            SelectOperator finalSelOp = pair.getFirst();</span><br><span class="line">            org.apache.hadoop.hive.ql.metadata.Table t = pair.getSecond();</span><br><span class="line">            List&lt;String&gt; colNames = <span class="keyword">null</span>;</span><br><span class="line">            <span class="keyword">if</span> (t != <span class="keyword">null</span>) &#123;</span><br><span class="line">                fieldSchemas = t.getCols();</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                <span class="comment">// Based on the plan outputs, find out the target table name and column names.</span></span><br><span class="line">                <span class="keyword">for</span> (WriteEntity output : plan.getOutputs()) &#123;</span><br><span class="line">                    Entity.Type entityType = output.getType();</span><br><span class="line">                    <span class="keyword">if</span> (entityType == Entity.Type.TABLE</span><br><span class="line">                            || entityType == Entity.Type.PARTITION) &#123;</span><br><span class="line">                        t = output.getTable();</span><br><span class="line">                        fieldSchemas = t.getCols();</span><br><span class="line">                        List&lt;FieldSchema&gt; cols = t.getCols();</span><br><span class="line">                        <span class="keyword">if</span> (cols != <span class="keyword">null</span> &amp;&amp; !cols.isEmpty()) &#123;</span><br><span class="line">                            colNames = Utilities.getColumnNamesFromFieldSchema(cols);</span><br><span class="line">                        &#125;</span><br><span class="line">                        <span class="keyword">break</span>;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">            Map&lt;ColumnInfo, LineageInfo.Dependency&gt; colMap = index.getDependencies(finalSelOp);</span><br><span class="line">            List&lt;LineageInfo.Dependency&gt; dependencies = colMap != <span class="keyword">null</span> ? Lists.newArrayList(colMap.values()) : <span class="keyword">null</span>;</span><br><span class="line">            <span class="keyword">int</span> fields = fieldSchemas.size();</span><br><span class="line">            <span class="keyword">if</span> (t != <span class="keyword">null</span> &amp;&amp; colMap != <span class="keyword">null</span> &amp;&amp; fields &lt; colMap.size()) &#123;</span><br><span class="line">                <span class="comment">// Dynamic partition keys should be added to field schemas.</span></span><br><span class="line">                List&lt;FieldSchema&gt; partitionKeys = t.getPartitionKeys();</span><br><span class="line">                <span class="keyword">int</span> dynamicKeyCount = colMap.size() - fields;</span><br><span class="line">                <span class="keyword">int</span> keyOffset = partitionKeys.size() - dynamicKeyCount;</span><br><span class="line">                <span class="keyword">if</span> (keyOffset &gt;= <span class="number">0</span>) &#123;</span><br><span class="line">                    fields += dynamicKeyCount;</span><br><span class="line">                    <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; dynamicKeyCount; i++) &#123;</span><br><span class="line">                        FieldSchema field = partitionKeys.get(keyOffset + i);</span><br><span class="line">                        fieldSchemas.add(field);</span><br><span class="line">                        <span class="keyword">if</span> (colNames != <span class="keyword">null</span>) &#123;</span><br><span class="line">                            colNames.add(field.getName());</span><br><span class="line">                        &#125;</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">            <span class="keyword">if</span> (dependencies == <span class="keyword">null</span> || dependencies.size() != fields) &#123;</span><br><span class="line">                log(<span class="string">"Result schema has "</span> + fields</span><br><span class="line">                        + <span class="string">" fields, but we don't get as many dependencies"</span>);</span><br><span class="line">            &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                <span class="comment">// Go through each target column, generate the lineage edges.</span></span><br><span class="line">                Set&lt;Vertex&gt; targets = <span class="keyword">new</span> LinkedHashSet&lt;Vertex&gt;();</span><br><span class="line">                <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; fields; i++) &#123;</span><br><span class="line">                    Vertex target = getOrCreateVertex(vertexCache,</span><br><span class="line">                            t.getTTable(), fieldSchemas.get(i),</span><br><span class="line">                            Vertex.Type.COLUMN);</span><br><span class="line"></span><br><span class="line">                    targets.add(target);</span><br><span class="line"></span><br><span class="line">                    LineageInfo.Dependency dep = dependencies.get(i);</span><br><span class="line">                    addEdge(vertexCache, edges, dep.getBaseCols(), target,</span><br><span class="line">                            dep.getExpr(), plan.getOperation());</span><br><span class="line"></span><br><span class="line">                    Vertex tTarget = getOrCreateVertex(vertexCache,</span><br><span class="line">                            t.getTTable(), <span class="keyword">null</span>,</span><br><span class="line">                            Vertex.Type.TABLE);</span><br><span class="line">                    targets.add(target);</span><br><span class="line">                    <span class="keyword">for</span> (LineageInfo.BaseColumnInfo baseCol : dep.getBaseCols()) &#123;</span><br><span class="line">                        baseCol.setColumn(<span class="keyword">null</span>);</span><br><span class="line">                    &#125;</span><br><span class="line">                    addEdge(vertexCache, edges, dep.getBaseCols(), tTarget,</span><br><span class="line">                            plan.getQueryString(), plan.getOperation());</span><br><span class="line"></span><br><span class="line">                &#125;</span><br><span class="line">                Set&lt;LineageInfo.Predicate&gt; conds = index.getPredicates(finalSelOp);</span><br><span class="line">                <span class="keyword">if</span> (conds != <span class="keyword">null</span> &amp;&amp; !conds.isEmpty()) &#123;</span><br><span class="line">                    <span class="keyword">for</span> (LineageInfo.Predicate cond : conds) &#123;</span><br><span class="line">                        addEdge(vertexCache, edges, cond.getBaseCols(),</span><br><span class="line">                                <span class="keyword">new</span> LinkedHashSet&lt;Vertex&gt;(targets), cond.getExpr(), plan.getOperation());</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> edges;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">addEdge</span><span class="params">(Map&lt;String, Vertex&gt; vertexCache, List&lt;Edge&gt; edges,</span></span></span><br><span class="line"><span class="function"><span class="params">                         Set&lt;LineageInfo.BaseColumnInfo&gt; srcCols, Vertex target, String expr, HiveOperation operation)</span> </span>&#123;</span><br><span class="line">        Set&lt;Vertex&gt; targets = <span class="keyword">new</span> LinkedHashSet&lt;Vertex&gt;();</span><br><span class="line">        targets.add(target);</span><br><span class="line">        addEdge(vertexCache, edges, srcCols, targets, expr, operation);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Find an edge from all edges that has the same source vertices.</span></span><br><span class="line"><span class="comment">     * If found, add the more targets to this edge's target vertex list.</span></span><br><span class="line"><span class="comment">     * Otherwise, create a new edge and add to edge list.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">addEdge</span><span class="params">(Map&lt;String, Vertex&gt; vertexCache, List&lt;Edge&gt; edges,</span></span></span><br><span class="line"><span class="function"><span class="params">                         Set&lt;LineageInfo.BaseColumnInfo&gt; srcCols, Set&lt;Vertex&gt; targets, String expr, HiveOperation operation)</span> </span>&#123;</span><br><span class="line">        Set&lt;Vertex&gt; sources = createSourceVertices(vertexCache, srcCols);</span><br><span class="line">        Edge edge = findSimilarEdgeBySources(edges, sources, expr, operation);</span><br><span class="line">        <span class="keyword">if</span> (edge == <span class="keyword">null</span>) &#123;</span><br><span class="line">            edges.add(<span class="keyword">new</span> Edge(sources, targets, expr, operation));</span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">            edge.targets.addAll(targets);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Convert a list of columns to a set of vertices.</span></span><br><span class="line"><span class="comment">     * Use cached vertices if possible.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> Set&lt;Vertex&gt; <span class="title">createSourceVertices</span><span class="params">(</span></span></span><br><span class="line"><span class="function"><span class="params">            Map&lt;String, Vertex&gt; vertexCache, Collection&lt;LineageInfo.BaseColumnInfo&gt; baseCols)</span> </span>&#123;</span><br><span class="line">        Set&lt;Vertex&gt; sources = <span class="keyword">new</span> LinkedHashSet&lt;Vertex&gt;();</span><br><span class="line">        <span class="keyword">if</span> (baseCols != <span class="keyword">null</span> &amp;&amp; !baseCols.isEmpty()) &#123;</span><br><span class="line">            <span class="keyword">for</span> (LineageInfo.BaseColumnInfo col : baseCols) &#123;</span><br><span class="line">                org.apache.hadoop.hive.metastore.api.Table table = col.getTabAlias().getTable();</span><br><span class="line">                <span class="keyword">if</span> (table.isTemporary()) &#123;</span><br><span class="line">                    <span class="comment">// Ignore temporary tables</span></span><br><span class="line">                    <span class="keyword">continue</span>;</span><br><span class="line">                &#125;</span><br><span class="line">                Vertex.Type type = Vertex.Type.TABLE;</span><br><span class="line">                FieldSchema fieldSchema = col.getColumn();</span><br><span class="line">                <span class="keyword">if</span> (fieldSchema != <span class="keyword">null</span>) &#123;</span><br><span class="line">                    type = Vertex.Type.COLUMN;</span><br><span class="line">                &#125;</span><br><span class="line"></span><br><span class="line">                sources.add(getOrCreateVertex(vertexCache, table, fieldSchema, type));</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> sources;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Find a vertex from a cache, or create one if not.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> Vertex <span class="title">getOrCreateVertex</span><span class="params">(</span></span></span><br><span class="line"><span class="function"><span class="params">            Map&lt;String, Vertex&gt; vertices, org.apache.hadoop.hive.metastore.api.Table t, FieldSchema fieldSchema, Vertex.Type type)</span> </span>&#123;</span><br><span class="line">        Vertex vertex;</span><br><span class="line">        <span class="keyword">if</span> (type.equals(Vertex.Type.TABLE)) &#123;</span><br><span class="line">            String key = getMd5Hash(type.name() + <span class="string">"."</span> + t.getDbName() + <span class="string">"."</span> + t.getTableName());</span><br><span class="line">            vertex = vertices.get(key);</span><br><span class="line">            <span class="keyword">if</span> (vertex == <span class="keyword">null</span>) &#123;</span><br><span class="line">                vertex = <span class="keyword">new</span> Vertex(type, DateFormatUtils.format(<span class="keyword">new</span> Date(t.getCreateTime() * <span class="number">1000L</span>), DATE_FORMAT), t.getDbName(),</span><br><span class="line">                        t.getOwner(), t.getTableName(), t.getTableType(), Vertex.Status.ONLINE, key);</span><br><span class="line">                vertices.put(key, vertex);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">            String key = getMd5Hash(type.name() + <span class="string">"."</span> + t.getDbName() + <span class="string">"."</span> + t.getTableName() + <span class="string">"."</span> + fieldSchema.getName());</span><br><span class="line">            vertex = vertices.get(key);</span><br><span class="line">            <span class="keyword">if</span> (vertex == <span class="keyword">null</span>) &#123;</span><br><span class="line">                vertex = <span class="keyword">new</span> Vertex(type, DateFormatUtils.format(<span class="keyword">new</span> Date(t.getCreateTime() * <span class="number">1000L</span>), DATE_FORMAT), t.getDbName(),</span><br><span class="line">                        t.getOwner(), t.getTableName(), t.getTableType(), fieldSchema.getName(), fieldSchema.getType(), Vertex.Status.ONLINE, key);</span><br><span class="line">                vertices.put(key, vertex);</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> vertex;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Find an edge that has the same type, expression, and sources.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> Edge <span class="title">findSimilarEdgeBySources</span><span class="params">(</span></span></span><br><span class="line"><span class="function"><span class="params">            List&lt;Edge&gt; edges, Set&lt;Vertex&gt; sources, String expr, HiveOperation operation)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">for</span> (Edge edge : edges) &#123;</span><br><span class="line">            <span class="keyword">if</span> (edge.operation == operation &amp;&amp; StringUtils.equals(edge.expr, expr)</span><br><span class="line">                    &amp;&amp; SetUtils.isEqualSet(edge.sources, sources)) &#123;</span><br><span class="line">                <span class="keyword">return</span> edge;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">null</span>;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Get all the vertices of all edges. Targets at first,</span></span><br><span class="line"><span class="comment">     * then sources. Assign id to each vertex.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> Set&lt;Vertex&gt; <span class="title">getVertices</span><span class="params">(List&lt;Edge&gt; edges)</span> </span>&#123;</span><br><span class="line">        Set&lt;Vertex&gt; vertices = <span class="keyword">new</span> LinkedHashSet&lt;Vertex&gt;();</span><br><span class="line">        <span class="keyword">for</span> (Edge edge : edges) &#123;</span><br><span class="line">            vertices.addAll(edge.targets);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">for</span> (Edge edge : edges) &#123;</span><br><span class="line">            vertices.addAll(edge.sources);</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="keyword">return</span> vertices;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Generate query string md5 hash.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> String <span class="title">getMd5Hash</span><span class="params">(String key)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> DigestUtils.md5Hex(key);</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Write out an JSON array of edges.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">writeEdges</span><span class="params">(JsonWriter writer, List&lt;SQLParse2.Edge&gt; edges)</span> <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">        writer.name(<span class="string">"edges"</span>);</span><br><span class="line">        writer.beginArray();</span><br><span class="line">        <span class="keyword">for</span> (SQLParse2.Edge edge : edges) &#123;</span><br><span class="line">            writer.beginObject();</span><br><span class="line">            writer.name(<span class="string">"sources"</span>);</span><br><span class="line">            writer.beginArray();</span><br><span class="line">            <span class="keyword">for</span> (SQLParse2.Vertex vertex : edge.sources) &#123;</span><br><span class="line">                writer.value(vertex.md5);</span><br><span class="line">            &#125;</span><br><span class="line">            writer.endArray();</span><br><span class="line">            writer.name(<span class="string">"targets"</span>);</span><br><span class="line">            writer.beginArray();</span><br><span class="line">            <span class="keyword">for</span> (SQLParse2.Vertex vertex : edge.targets) &#123;</span><br><span class="line">                writer.value(vertex.md5);</span><br><span class="line">            &#125;</span><br><span class="line">            writer.endArray();</span><br><span class="line">            <span class="keyword">if</span> (edge.expr != <span class="keyword">null</span>) &#123;</span><br><span class="line">                writer.name(<span class="string">"expr"</span>).value(edge.expr);</span><br><span class="line">            &#125;</span><br><span class="line">            writer.name(<span class="string">"operation"</span>).value(edge.operation.name());</span><br><span class="line">            writer.endObject();</span><br><span class="line">        &#125;</span><br><span class="line">        writer.endArray();</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="comment">/**</span></span><br><span class="line"><span class="comment">     * Write out an JSON array of vertices.</span></span><br><span class="line"><span class="comment">     */</span></span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">writeVertices</span><span class="params">(JsonWriter writer, Set&lt;SQLParse2.Vertex&gt; vertices)</span> <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">        writer.name(<span class="string">"vertices"</span>);</span><br><span class="line">        writer.beginArray();</span><br><span class="line">        <span class="keyword">for</span> (SQLParse2.Vertex vertex : vertices) &#123;</span><br><span class="line">            writer.beginObject();</span><br><span class="line">            writer.name(<span class="string">"md5"</span>).value(vertex.md5);</span><br><span class="line">            writer.name(<span class="string">"type"</span>).value(vertex.type.name());</span><br><span class="line">            writer.name(<span class="string">"createAt"</span>).value(vertex.createAt);</span><br><span class="line">            writer.name(<span class="string">"db"</span>).value(vertex.db);</span><br><span class="line">            writer.name(<span class="string">"owner"</span>).value(vertex.owner);</span><br><span class="line">            writer.name(<span class="string">"table"</span>).value(vertex.table);</span><br><span class="line">            writer.name(<span class="string">"tableType"</span>).value(vertex.tableType);</span><br><span class="line">            writer.name(<span class="string">"col"</span>).value(vertex.col);</span><br><span class="line">            writer.name(<span class="string">"colType"</span>).value(vertex.colType);</span><br><span class="line">            writer.name(<span class="string">"status"</span>).value(vertex.status.name());</span><br><span class="line">            writer.endObject();</span><br><span class="line">        &#125;</span><br><span class="line">        writer.endArray();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<blockquote>
<p>注意事项</p>
</blockquote>
<ul>
<li>开启血缘信息：查看hive源码发现hive的hive.exec.post.hooks配置需要包含org.apache.hadoop.hive.ql.hooks.PostExecutePrinter、org.apache.hadoop.hive.ql.hooks.LineageLogger或org.apache.atlas.hive.hook.HiveHook中任意一个Hook函数，SessionState中才会有血缘关系。因此可以在配置中加入其中任意一个Hook配置：conf.setVar(HiveConf.ConfVars.POSTEXECHOOKS, “org.apache.hadoop.hive.ql.hooks.PostExecutePrinter”)。</li>
</ul>
<p>相关代码可以参见SemanticAnalyzer类：</p>
<p><img src="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210903170253930.png" alt="image-20210903170253930"></p>
<ul>
<li>过滤表/视图存在与否检测：修改SemanticAnalyzer中检查表/视图存在与否的代码，重新打包，使SQL解析工具使用这个hive-exec-2.3.7.jar这个包。不然SQL解析过程中会报错。</li>
<li>开启动态分区非严格模式：conf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, “nonstrict”)</li>
</ul>
<h4 id="Flink任务"><a href="#Flink任务" class="headerlink" title="Flink任务"></a>Flink任务</h4><p>有了SQL解析工具后，就可以编写Flink程序消费Kafka数据解析SQL，然后录入JanusGraph。本文只提供录入JanusGraph方案，录入ElasticSearch请自行实现。</p>
<h5 id="LineageTask"><a href="#LineageTask" class="headerlink" title="LineageTask"></a>LineageTask</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.task.service</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.time.Duration</span><br><span class="line"><span class="keyword">import</span> java.util.Properties</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.core.bean.Constants</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.core.util.&#123;EnvUtil, PropertiesUtil&#125;</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.task.function.&#123;LineageBuildMapFunction, LineageProcessWindowFunction&#125;</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.task.sink.LineageToGraphSinkFunction</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.api.common.JobExecutionResult</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.api.common.eventtime.&#123;SerializableTimestampAssigner, WatermarkStrategy&#125;</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.api.common.functions.RichMapFunction</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.api.common.serialization.SimpleStringSchema</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.api.scala._</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.configuration.Configuration</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.TimeCharacteristic</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.scala.OutputTag</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.windowing.time.Time</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer</span><br><span class="line"><span class="keyword">import</span> org.codehaus.jackson.map.ObjectMapper</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@date</span> 2021-08-17 14:07</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@description</span> </span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line"><span class="class"><span class="keyword">class</span> <span class="title">LineageTask</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">buildGraph</span><span class="params">(topic: String, configurationFile: String, remoteTraversalSourceName: String)</span>: JobExecutionResult </span>= &#123;</span><br><span class="line">    val env = EnvUtil.getFlinkStreamEnv()</span><br><span class="line">    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)</span><br><span class="line">    env.setParallelism(<span class="number">1</span>)</span><br><span class="line"></span><br><span class="line">    val properties = <span class="keyword">new</span> Properties()</span><br><span class="line">    properties.setProperty(Constants.KAFKA_BOOTSTRAP_SERVERS, PropertiesUtil.getValue(Constants.KAFKA_BOOTSTRAP_SERVERS))</span><br><span class="line">    properties.setProperty(Constants.KAFKA_GROUP_ID, PropertiesUtil.getValue(Constants.KAFKA_GROUP_ID))</span><br><span class="line"></span><br><span class="line">    val consumer = <span class="keyword">new</span> FlinkKafkaConsumer[String](topic, <span class="keyword">new</span> SimpleStringSchema(), properties)</span><br><span class="line">    val consumerStream = env.addSource(consumer)</span><br><span class="line">      .map(<span class="keyword">new</span> RichMapFunction[String, SqlEntity] &#123;</span><br><span class="line">        <span class="keyword">private</span> <span class="keyword">var</span> MAPPER: ObjectMapper = _</span><br><span class="line"></span><br><span class="line">        <span class="function">override def <span class="title">open</span><span class="params">(parameters: Configuration)</span>: Unit </span>= &#123;</span><br><span class="line">          MAPPER = <span class="keyword">new</span> ObjectMapper</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function">override def <span class="title">map</span><span class="params">(element: String)</span>: SqlEntity </span>= &#123;</span><br><span class="line">          MAPPER.readValue[SqlEntity](element, classOf[SqlEntity])</span><br><span class="line">        &#125;</span><br><span class="line">      &#125;)</span><br><span class="line"></span><br><span class="line">    val watermarkStrategy = WatermarkStrategy</span><br><span class="line">      .forBoundedOutOfOrderness[SqlEntity](Duration.ofSeconds(<span class="number">10</span>))</span><br><span class="line">      .withTimestampAssigner(<span class="keyword">new</span> SerializableTimestampAssigner[SqlEntity] &#123;</span><br><span class="line">        <span class="function">override def <span class="title">extractTimestamp</span><span class="params">(sqlEntity: SqlEntity, l: Long)</span>: Long </span>= &#123;</span><br><span class="line">          println(sqlEntity.getTimestamp)</span><br><span class="line">          sqlEntity.getTimestamp</span><br><span class="line">        &#125;</span><br><span class="line">      &#125;)</span><br><span class="line"></span><br><span class="line">    val watermarkStream = consumerStream.assignTimestampsAndWatermarks(watermarkStrategy)</span><br><span class="line"></span><br><span class="line">    val outPutTag = <span class="keyword">new</span> OutputTag[SqlEntity](<span class="string">"late-data"</span>) &#123;&#125;</span><br><span class="line">    val lineageStream = watermarkStream.</span><br><span class="line">      keyBy(_.getDatabase)</span><br><span class="line">      .timeWindow(Time.seconds(<span class="number">20</span>))</span><br><span class="line">      .allowedLateness(Time.seconds(<span class="number">5</span>))</span><br><span class="line">      .sideOutputLateData(outPutTag)</span><br><span class="line">      .trigger(EventTimeTrigger.create())</span><br><span class="line">      .process(<span class="keyword">new</span> LineageProcessWindowFunction)</span><br><span class="line">      .map(<span class="keyword">new</span> LineageBuildMapFunction)</span><br><span class="line"></span><br><span class="line">    lineageStream.addSink(<span class="keyword">new</span> LineageToGraphSinkFunction(configurationFile, remoteTraversalSourceName))</span><br><span class="line">    <span class="comment">//获取迟到太久的数据</span></span><br><span class="line">    val sideOutput = watermarkStream.getSideOutput[SqlEntity](outPutTag)</span><br><span class="line">    val sideOutPutStream = sideOutput.map(<span class="keyword">new</span> LineageBuildMapFunction)</span><br><span class="line">    sideOutPutStream.addSink(<span class="keyword">new</span> LineageToGraphSinkFunction(configurationFile, remoteTraversalSourceName))</span><br><span class="line"></span><br><span class="line">    env.execute()</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">object LineageTask &#123;</span><br><span class="line">  <span class="function">def <span class="title">main</span><span class="params">(args: Array[String])</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">new</span> LineageTask().buildGraph(<span class="string">"lineage_sql"</span>, <span class="string">"remote.yaml"</span>, <span class="string">"hive_traversal"</span>)</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h5 id="SqlEntity"><a href="#SqlEntity" class="headerlink" title="SqlEntity"></a>SqlEntity</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br><span class="line">157</span><br><span class="line">158</span><br><span class="line">159</span><br><span class="line">160</span><br><span class="line">161</span><br><span class="line">162</span><br><span class="line">163</span><br><span class="line">164</span><br><span class="line">165</span><br><span class="line">166</span><br><span class="line">167</span><br><span class="line">168</span><br><span class="line">169</span><br><span class="line">170</span><br><span class="line">171</span><br><span class="line">172</span><br><span class="line">173</span><br><span class="line">174</span><br><span class="line">175</span><br><span class="line">176</span><br><span class="line">177</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.hive.core;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.io.Serializable;</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@date</span> 2021-08-17 10:43</span></span><br><span class="line"><span class="comment"> * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment"> */</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SqlEntity</span> <span class="keyword">implements</span> <span class="title">Serializable</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> String logtime;</span><br><span class="line">    <span class="keyword">private</span> String sql;</span><br><span class="line">    <span class="keyword">private</span> String user;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">long</span> timestamp;</span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">long</span> duration;</span><br><span class="line">    <span class="keyword">private</span> String jobids;</span><br><span class="line">    <span class="keyword">private</span> String engine;</span><br><span class="line">    <span class="keyword">private</span> String database;</span><br><span class="line">    <span class="keyword">private</span> String datasource;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="title">SqlEntity</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="title">SqlEntity</span><span class="params">(String logtime, String sql, String user, <span class="keyword">long</span> timestamp, <span class="keyword">long</span> duration, String jobids, String engine, String database, String datasource)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.logtime = logtime;</span><br><span class="line">        <span class="keyword">this</span>.sql = sql;</span><br><span class="line">        <span class="keyword">this</span>.user = user;</span><br><span class="line">        <span class="keyword">this</span>.timestamp = timestamp;</span><br><span class="line">        <span class="keyword">this</span>.duration = duration;</span><br><span class="line">        <span class="keyword">this</span>.jobids = jobids;</span><br><span class="line">        <span class="keyword">this</span>.engine = engine;</span><br><span class="line">        <span class="keyword">this</span>.database = database;</span><br><span class="line">        <span class="keyword">this</span>.datasource = datasource;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> Builder <span class="title">newBuilder</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> Builder();</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">static</span> <span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Builder</span> </span>&#123;</span><br><span class="line">        <span class="keyword">private</span> String logtime;</span><br><span class="line">        <span class="keyword">private</span> String sql;</span><br><span class="line">        <span class="keyword">private</span> String user;</span><br><span class="line">        <span class="keyword">private</span> <span class="keyword">long</span> timestamp;</span><br><span class="line">        <span class="keyword">private</span> <span class="keyword">long</span> duration;</span><br><span class="line">        <span class="keyword">private</span> String jobids;</span><br><span class="line">        <span class="keyword">private</span> String engine;</span><br><span class="line">        <span class="keyword">private</span> String database;</span><br><span class="line">        <span class="keyword">private</span> String datasource;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">private</span> <span class="title">Builder</span><span class="params">()</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">logtime</span><span class="params">(String logtime)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.logtime = logtime;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">sql</span><span class="params">(String sql)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.sql = sql;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">user</span><span class="params">(String user)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.user = user;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">timestamp</span><span class="params">(<span class="keyword">long</span> timestamp)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.timestamp = timestamp;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">duration</span><span class="params">(<span class="keyword">long</span> duration)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.duration = duration;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">jobids</span><span class="params">(String jobids)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.jobids = jobids;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">engine</span><span class="params">(String engine)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.engine = engine;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">database</span><span class="params">(String database)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.database = database;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> Builder <span class="title">datasource</span><span class="params">(String datasource)</span> </span>&#123;</span><br><span class="line">            <span class="keyword">this</span>.datasource = datasource;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">this</span>;</span><br><span class="line">        &#125;</span><br><span class="line"></span><br><span class="line">        <span class="function"><span class="keyword">public</span> SqlEntity <span class="title">build</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">return</span> <span class="keyword">new</span> SqlEntity(<span class="keyword">this</span>.logtime, <span class="keyword">this</span>.sql, <span class="keyword">this</span>.user, <span class="keyword">this</span>.timestamp, <span class="keyword">this</span>.duration</span><br><span class="line">                    , <span class="keyword">this</span>.jobids, <span class="keyword">this</span>.engine, <span class="keyword">this</span>.database, <span class="keyword">this</span>.datasource);</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getLogtime</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> logtime;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setLogtime</span><span class="params">(String logtime)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.logtime = logtime;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getSql</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> sql;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setSql</span><span class="params">(String sql)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.sql = sql;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getUser</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> user;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setUser</span><span class="params">(String user)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.user = user;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">long</span> <span class="title">getTimestamp</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> timestamp;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setTimestamp</span><span class="params">(<span class="keyword">long</span> timestamp)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.timestamp = timestamp;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">long</span> <span class="title">getDuration</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> duration;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setDuration</span><span class="params">(<span class="keyword">long</span> duration)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.duration = duration;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getJobids</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> jobids;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setJobids</span><span class="params">(String jobids)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.jobids = jobids;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getEngine</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> engine;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setEngine</span><span class="params">(String engine)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.engine = engine;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getDatabase</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> database;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setDatabase</span><span class="params">(String database)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.database = database;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> String <span class="title">getDatasource</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        <span class="keyword">return</span> datasource;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setDatasource</span><span class="params">(String datasource)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">this</span>.datasource = datasource;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h5 id="LineageProcessWindowFunction"><a href="#LineageProcessWindowFunction" class="headerlink" title="LineageProcessWindowFunction"></a>LineageProcessWindowFunction</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.task.function</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.scala.function.ProcessWindowFunction</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.windowing.windows.TimeWindow</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.util.Collector</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> scala.collection.mutable</span><br><span class="line"><span class="keyword">import</span> scala.collection.mutable.ArrayBuffer</span><br><span class="line"><span class="keyword">import</span> scala.util.Sorting</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@date</span> 2021-08-17 15:34</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line">class LineageProcessWindowFunction() extends ProcessWindowFunction[SqlEntity, SqlEntity, String, TimeWindow] &#123;</span><br><span class="line">  <span class="function">override def <span class="title">process</span><span class="params">(key: String, context: Context, elements: Iterable[SqlEntity], out: Collector[SqlEntity])</span>: Unit </span>= &#123;</span><br><span class="line">    val arrBuf = ArrayBuffer[Long]()</span><br><span class="line">    val map = <span class="keyword">new</span> mutable.HashMap[Long, SqlEntity]()</span><br><span class="line"></span><br><span class="line">    <span class="comment">//组装结果</span></span><br><span class="line">    elements.map(line =&gt; &#123;</span><br><span class="line">      arrBuf.append(line.getTimestamp)</span><br><span class="line">      map.put(line.getTimestamp, line)</span><br><span class="line">    &#125;)</span><br><span class="line"></span><br><span class="line">    val arr = arrBuf.toArray</span><br><span class="line">    <span class="comment">//对窗口内的数据按照时间进行排序</span></span><br><span class="line">    Sorting.quickSort(arr)</span><br><span class="line">    val arrayLength = arr.length</span><br><span class="line">    <span class="keyword">for</span> (i &lt;- <span class="number">0</span> until arrayLength) &#123;</span><br><span class="line">      val timestamp = arr(i)</span><br><span class="line">      val sqlEntity = map(timestamp)</span><br><span class="line"></span><br><span class="line">      out.collect(sqlEntity)</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h5 id="LineageBuildMapFunction"><a href="#LineageBuildMapFunction" class="headerlink" title="LineageBuildMapFunction"></a>LineageBuildMapFunction</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.task.function</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.SQLParse2</span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.core.SqlEntity</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.api.common.functions.RichMapFunction</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.configuration.Configuration</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@date</span> 2021-08-17 16:18</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line">class LineageBuildMapFunction extends RichMapFunction[SqlEntity, java.util.List[SQLParse2.Edge]]&#123;</span><br><span class="line">  <span class="keyword">var</span> sqlParse2: SQLParse2 = _</span><br><span class="line"></span><br><span class="line">  <span class="function">override def <span class="title">open</span><span class="params">(parameters: Configuration)</span>: Unit </span>= &#123;</span><br><span class="line">    sqlParse2 = <span class="keyword">new</span> SQLParse2</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">override def <span class="title">map</span><span class="params">(sqlEntity: SqlEntity)</span>: util.List[SQLParse2.Edge] </span>= &#123;</span><br><span class="line">    sqlParse2.run(sqlEntity.getSql)</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h5 id="LineageToGraphSinkFunction"><a href="#LineageToGraphSinkFunction" class="headerlink" title="LineageToGraphSinkFunction"></a>LineageToGraphSinkFunction</h5><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br><span class="line">157</span><br><span class="line">158</span><br><span class="line">159</span><br><span class="line">160</span><br><span class="line">161</span><br><span class="line">162</span><br><span class="line">163</span><br><span class="line">164</span><br><span class="line">165</span><br><span class="line">166</span><br><span class="line">167</span><br><span class="line">168</span><br><span class="line">169</span><br><span class="line">170</span><br><span class="line">171</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> com.xyueji.traffic.task.sink</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.hive.SQLParse2</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.lang3.StringUtils</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.configuration</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.streaming.api.functions.sink.RichSinkFunction</span><br><span class="line"><span class="keyword">import</span> org.apache.flink.util.SerializableObject</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.driver.Cluster</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.driver.remote.DriverRemoteConnection</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.structure.&#123;Graph, Vertex&#125;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> scala.collection.JavaConversions._</span><br><span class="line"><span class="keyword">import</span> scala.util.control.Breaks._</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@date</span> 2021-08-17 16:18</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line">class LineageToGraphSinkFunction(configurationFile: String, remoteTraversalSourceName: String, maxNumRetries: Int = 3) extends RichSinkFunction[java.util.List[SQLParse2.Edge]] &#123;</span><br><span class="line">  <span class="keyword">var</span> cluster: Cluster = _</span><br><span class="line">  <span class="keyword">var</span> graph: Graph = _</span><br><span class="line">  <span class="keyword">var</span> g: GraphTraversalSource = _</span><br><span class="line">  val lock: SerializableObject = <span class="keyword">new</span> SerializableObject</span><br><span class="line">  <span class="keyword">var</span> isRunning: Boolean = <span class="keyword">true</span></span><br><span class="line">  <span class="keyword">var</span> retries: Int = <span class="number">0</span></span><br><span class="line"></span><br><span class="line">  <span class="comment">/**</span></span><br><span class="line"><span class="comment">    * 创建连接</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">  <span class="function">def <span class="title">createConnection</span><span class="params">()</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="comment">//连接Gremlin Server</span></span><br><span class="line">    cluster = Cluster.open(configurationFile)</span><br><span class="line"></span><br><span class="line">    graph = EmptyGraph.instance</span><br><span class="line">    g = graph.traversal.withRemote(DriverRemoteConnection.using(cluster, remoteTraversalSourceName))</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">override def <span class="title">open</span><span class="params">(parameters: configuration.Configuration)</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">try</span></span><br><span class="line">      <span class="keyword">this</span>.lock <span class="keyword">synchronized</span> <span class="keyword">this</span>.createConnection()</span><br><span class="line"></span><br><span class="line">    <span class="keyword">catch</span> &#123;</span><br><span class="line">      <span class="keyword">case</span> var5: Exception =&gt;</span><br><span class="line">        <span class="keyword">throw</span> <span class="keyword">new</span> Exception(<span class="string">"Cannot connect to gremlin server"</span>, var5)</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">override def <span class="title">invoke</span><span class="params">(edges: util.List[SQLParse2.Edge])</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">      addElements(edges)</span><br><span class="line">    &#125; <span class="keyword">catch</span> &#123;</span><br><span class="line">      <span class="keyword">case</span> var12: Exception =&gt;</span><br><span class="line">        <span class="keyword">if</span> (<span class="keyword">this</span>.maxNumRetries == <span class="number">0</span>) <span class="keyword">throw</span> <span class="keyword">new</span> Exception(<span class="string">"Failed to insert elements to graph. Connection re-tries are not enabled."</span>, var12)</span><br><span class="line">        <span class="keyword">this</span>.lock <span class="keyword">synchronized</span> &#123;</span><br><span class="line">          <span class="keyword">var</span> lastException: Exception = <span class="keyword">null</span></span><br><span class="line">          <span class="keyword">this</span>.retries = <span class="number">0</span></span><br><span class="line"></span><br><span class="line">          breakable &#123;</span><br><span class="line">            <span class="keyword">while</span> (<span class="keyword">this</span>.isRunning) &#123;</span><br><span class="line">              <span class="keyword">if</span> (!<span class="keyword">this</span>.isRunning || <span class="keyword">this</span>.maxNumRetries &gt;= <span class="number">0</span> &amp;&amp; <span class="keyword">this</span>.retries &gt;= <span class="keyword">this</span>.maxNumRetries) &#123;</span><br><span class="line">                <span class="keyword">if</span> (<span class="keyword">this</span>.isRunning) <span class="keyword">throw</span> <span class="keyword">new</span> Exception(<span class="string">"Failed to insert elements to graph. Failed after "</span> + <span class="keyword">this</span>.retries + <span class="string">" retries."</span>, lastException)</span><br><span class="line">                <span class="keyword">this</span>.isRunning = <span class="keyword">false</span></span><br><span class="line">                <span class="keyword">break</span>() <span class="comment">//todo: break is not supported</span></span><br><span class="line">              &#125;</span><br><span class="line">              <span class="keyword">try</span></span><br><span class="line">                  <span class="keyword">if</span> (g != <span class="keyword">null</span>) g.close()</span><br><span class="line">              <span class="keyword">catch</span> &#123;</span><br><span class="line">                <span class="keyword">case</span> var9: Exception =&gt;</span><br><span class="line">                  println(<span class="string">"Could not close g"</span>, var9)</span><br><span class="line">              &#125;</span><br><span class="line">              <span class="keyword">try</span></span><br><span class="line">                  <span class="keyword">if</span> (cluster != <span class="keyword">null</span>) cluster.close()</span><br><span class="line">              <span class="keyword">catch</span> &#123;</span><br><span class="line">                <span class="keyword">case</span> var8: Exception =&gt;</span><br><span class="line">                  println(<span class="string">"Could not close cluster"</span>, var8)</span><br><span class="line">              &#125;</span><br><span class="line">              graph = <span class="keyword">null</span></span><br><span class="line">              <span class="keyword">this</span>.retries += <span class="number">1</span></span><br><span class="line">              <span class="keyword">try</span> &#123;</span><br><span class="line">                <span class="keyword">this</span>.createConnection()</span><br><span class="line">                addElements(edges)</span><br><span class="line">              &#125; <span class="keyword">catch</span> &#123;</span><br><span class="line">                <span class="keyword">case</span> var10: Exception =&gt;</span><br><span class="line">                  lastException = var10</span><br><span class="line">                  println(<span class="string">"Re-connect to gremlin server failed. Retry time(s): "</span> + <span class="keyword">this</span>.retries, var10)</span><br><span class="line">                  <span class="keyword">this</span>.lock.wait(<span class="number">500L</span>)</span><br><span class="line">              &#125;</span><br><span class="line">            &#125;</span><br><span class="line">          &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">addElements</span><span class="params">(edges: util.List[SQLParse2.Edge])</span>: Unit </span>= &#123;</span><br><span class="line"></span><br><span class="line">    <span class="keyword">for</span> (edge &lt;- edges) &#123;</span><br><span class="line">      <span class="keyword">for</span> (source &lt;- edge.getSources) &#123;</span><br><span class="line">        val fromV: Vertex = getVertex(source)</span><br><span class="line">        <span class="keyword">for</span> (target &lt;- edge.getTargets) &#123;</span><br><span class="line">          val toV: Vertex = getVertex(target)</span><br><span class="line"></span><br><span class="line">          buildEdge(edge, fromV, toV)</span><br><span class="line">        &#125;</span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">buildEdge</span><span class="params">(e: SQLParse2.Edge, fromV: Vertex, toV: Vertex)</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">if</span> (StringUtils.isNotEmpty(e.getExpr)) &#123;</span><br><span class="line">      g.addE(e.getOperation.getOperationName).from(fromV).to(toV).property(<span class="string">"expr"</span>, e.getExpr).next()</span><br><span class="line">    &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">      g.addE(e.getOperation.getOperationName).from(fromV).to(toV).next()</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getVertex</span><span class="params">(v: SQLParse2.Vertex)</span>: Vertex </span>= &#123;</span><br><span class="line">    val exist = g.V().has(<span class="string">"md5"</span>, v.getMd5).hasNext</span><br><span class="line">    <span class="keyword">if</span> (!exist) &#123;</span><br><span class="line">      v.getType match &#123;</span><br><span class="line">        <span class="keyword">case</span> SQLParse2.Vertex.Type.TABLE =&gt; <span class="keyword">return</span> g.addV(v.getType.name())</span><br><span class="line">          .property(<span class="string">"md5"</span>, v.getMd5)</span><br><span class="line">          .property(<span class="string">"createAt"</span>, v.getCreateAt)</span><br><span class="line">          .property(<span class="string">"db"</span>, v.getDb)</span><br><span class="line">          .property(<span class="string">"owner"</span>, v.getOwner)</span><br><span class="line">          .property(<span class="string">"table"</span>, v.getTable)</span><br><span class="line">          .property(<span class="string">"tableType"</span>, v.getTableType)</span><br><span class="line">          .property(<span class="string">"status"</span>, v.getStatus.name())</span><br><span class="line">          .next()</span><br><span class="line"></span><br><span class="line">        <span class="keyword">case</span> SQLParse2.Vertex.Type.COLUMN =&gt; <span class="keyword">return</span> g.addV(v.getType.name())</span><br><span class="line">          .property(<span class="string">"md5"</span>, v.getMd5)</span><br><span class="line">          .property(<span class="string">"createAt"</span>, v.getCreateAt)</span><br><span class="line">          .property(<span class="string">"db"</span>, v.getDb)</span><br><span class="line">          .property(<span class="string">"owner"</span>, v.getOwner)</span><br><span class="line">          .property(<span class="string">"table"</span>, v.getTable)</span><br><span class="line">          .property(<span class="string">"tableType"</span>, v.getTableType)</span><br><span class="line">          .property(<span class="string">"col"</span>, v.getCol)</span><br><span class="line">          .property(<span class="string">"colType"</span>, v.getColType)</span><br><span class="line">          .property(<span class="string">"status"</span>, v.getStatus.name())</span><br><span class="line">          .next()</span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    g.V().has(<span class="string">"md5"</span>, v.getMd5).next()</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">override def <span class="title">close</span><span class="params">()</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">this</span>.isRunning = <span class="keyword">false</span></span><br><span class="line">    <span class="keyword">this</span>.lock <span class="keyword">synchronized</span> &#123;</span><br><span class="line">      <span class="keyword">this</span>.lock.notifyAll()</span><br><span class="line"></span><br><span class="line">      <span class="keyword">try</span> &#123;</span><br><span class="line">        <span class="keyword">if</span> (g != <span class="keyword">null</span>) &#123; <span class="comment">// this closes the remote, no need to close the empty graph</span></span><br><span class="line">          g.close()</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (cluster != <span class="keyword">null</span>) &#123; <span class="comment">// the cluster closes all of its clients</span></span><br><span class="line">          cluster.close()</span><br><span class="line">        &#125;</span><br><span class="line">      &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">        g = <span class="keyword">null</span></span><br><span class="line">        graph = <span class="keyword">null</span></span><br><span class="line">        cluster = <span class="keyword">null</span></span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h5 id="remote-yaml"><a href="#remote-yaml" class="headerlink" title="remote.yaml"></a>remote.yaml</h5><figure class="highlight yaml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="attr">hosts:</span> <span class="string">[hadoop2]</span></span><br><span class="line"><span class="attr">port:</span> <span class="number">8182</span></span><br><span class="line"><span class="attr">username:</span> <span class="string">bigdata</span></span><br><span class="line"><span class="attr">password:</span> <span class="number">123456</span></span><br><span class="line"><span class="attr">serializer:</span> <span class="string">&#123;</span></span><br><span class="line"><span class="attr">  className:</span> <span class="string">org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV3d0,</span></span><br><span class="line"><span class="attr">  config:</span> <span class="string">&#123;</span></span><br><span class="line"><span class="attr">    ioRegistries:</span> <span class="string">[org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry]</span></span><br><span class="line">  <span class="string">&#125;</span></span><br><span class="line"><span class="string">&#125;</span></span><br></pre></td></tr></table></figure>
<h3 id="数据血缘查询"><a href="#数据血缘查询" class="headerlink" title="数据血缘查询"></a>数据血缘查询</h3><p>数据录入JanusGraph后就可以通过连接GremlinServer来查询血缘数据。本文编写一个spark程序例子，查询GremlinServer生成供Echarts使用的图json文件。</p>
<h4 id="GraphToJsonTask"><a href="#GraphToJsonTask" class="headerlink" title="GraphToJsonTask"></a>GraphToJsonTask</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br><span class="line">131</span><br><span class="line">132</span><br><span class="line">133</span><br><span class="line">134</span><br><span class="line">135</span><br><span class="line">136</span><br><span class="line">137</span><br><span class="line">138</span><br><span class="line">139</span><br><span class="line">140</span><br><span class="line">141</span><br><span class="line">142</span><br><span class="line">143</span><br><span class="line">144</span><br><span class="line">145</span><br><span class="line">146</span><br><span class="line">147</span><br><span class="line">148</span><br><span class="line">149</span><br><span class="line">150</span><br><span class="line">151</span><br><span class="line">152</span><br><span class="line">153</span><br><span class="line">154</span><br><span class="line">155</span><br><span class="line">156</span><br><span class="line">157</span><br><span class="line">158</span><br><span class="line">159</span><br><span class="line">160</span><br><span class="line">161</span><br><span class="line">162</span><br><span class="line">163</span><br><span class="line">164</span><br><span class="line">165</span><br><span class="line">166</span><br><span class="line">167</span><br><span class="line">168</span><br><span class="line">169</span><br><span class="line">170</span><br><span class="line">171</span><br><span class="line">172</span><br><span class="line">173</span><br><span class="line">174</span><br><span class="line">175</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> spark.com.xyueji.traffic.service</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> com.xyueji.traffic.core.util.EnvUtil</span><br><span class="line"><span class="keyword">import</span> org.apache.commons.lang3.StringUtils</span><br><span class="line"><span class="keyword">import</span> org.apache.spark.SparkConf</span><br><span class="line"><span class="keyword">import</span> org.apache.spark.sql.SaveMode</span><br><span class="line"><span class="keyword">import</span> spark.com.xyueji.traffic.bean.&#123;Categories, Edge, Vertex&#125;</span><br><span class="line"><span class="keyword">import</span> spark.com.xyueji.traffic.dao.GraphDao</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> scala.collection.JavaConverters._</span><br><span class="line"><span class="keyword">import</span> scala.util.Random</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@date</span> 2021-08-24 11:49</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@description</span></span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line"><span class="class"><span class="keyword">class</span> <span class="title">GraphToJsonTask</span> </span>&#123;</span><br><span class="line">  val graphDao: GraphDao = <span class="keyword">new</span> GraphDao(<span class="string">"remote.yaml"</span>, <span class="string">"hive_traversal"</span>)</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">save</span><span class="params">()</span>: Unit </span>= &#123;</span><br><span class="line">    val conf = <span class="keyword">new</span> SparkConf().setMaster(<span class="string">"local[*]"</span>).setAppName(<span class="string">"GraphToJson"</span>)</span><br><span class="line">    <span class="comment">// 设置序列化器为KryoSerializer。</span></span><br><span class="line">    conf.set(<span class="string">"spark.serializer"</span>, <span class="string">"org.apache.spark.serializer.KryoSerializer"</span>)</span><br><span class="line">    <span class="comment">// 注册要序列化的自定义类型。</span></span><br><span class="line">    conf.registerKryoClasses(Array(classOf[Vertex], classOf[Edge], classOf[Categories]))</span><br><span class="line"></span><br><span class="line">    val spark = EnvUtil.getSparkEnv(conf)</span><br><span class="line"></span><br><span class="line">    <span class="keyword">import</span> spark.implicits._</span><br><span class="line"></span><br><span class="line">    val sc = spark.sparkContext</span><br><span class="line"></span><br><span class="line">    val vertexes = graphDao.listAllV()</span><br><span class="line">    val labels = <span class="keyword">new</span> util.HashSet[String]</span><br><span class="line">    vertexes.foreach(v =&gt; &#123;</span><br><span class="line">      labels.add(v.get(<span class="string">"label"</span>).mkString)</span><br><span class="line">    &#125;)</span><br><span class="line">    val bothVertexes = graphDao.listAllBothVertex</span><br><span class="line"></span><br><span class="line">    graphDao.close()</span><br><span class="line">    val vertexesRdd = sc.parallelize(vertexes)</span><br><span class="line">    val labelList = labels.asScala.toList</span><br><span class="line">    val labelsRdd = sc.parallelize(labelList)</span><br><span class="line">    val bothVertexesRdd = sc.parallelize(bothVertexes)</span><br><span class="line"></span><br><span class="line">    val vertexesDF = vertexesRdd.map(v =&gt; &#123;</span><br><span class="line">      val id = v.get(<span class="string">"id"</span>).mkString</span><br><span class="line">      val db = v.get(<span class="string">"db"</span>).mkString</span><br><span class="line">      val table = v.get(<span class="string">"table"</span>).mkString</span><br><span class="line">      val owner = v.get(<span class="string">"owner"</span>).mkString</span><br><span class="line">      val tableType = v.get(<span class="string">"tableType"</span>).mkString</span><br><span class="line">      val col = v.get(<span class="string">"col"</span>).mkString</span><br><span class="line">      val colType = v.get(<span class="string">"colType"</span>).mkString</span><br><span class="line">      val status = v.get(<span class="string">"status"</span>).mkString</span><br><span class="line"></span><br><span class="line">      val category = v.get(<span class="string">"label"</span>).mkString</span><br><span class="line">      val createAt = v.get(<span class="string">"createAt"</span>).mkString</span><br><span class="line"></span><br><span class="line">      <span class="keyword">new</span> Tuple10[String, String, String, String, String, String, String, String, String, String](id, db, table, owner, tableType, col, colType, status, category, createAt)</span><br><span class="line">    &#125;).toDF(<span class="string">"id"</span>, <span class="string">"db"</span>, <span class="string">"table"</span>, <span class="string">"owner"</span>, <span class="string">"tableType"</span>, <span class="string">"col"</span>, <span class="string">"colType"</span>, <span class="string">"status"</span>, <span class="string">"category"</span>, <span class="string">"createAt"</span>)</span><br><span class="line"></span><br><span class="line">    vertexesDF.createOrReplaceTempView(<span class="string">"t_vertex"</span>)</span><br><span class="line"></span><br><span class="line">    val bothVertexesDF = bothVertexesRdd.map(bothV =&gt; &#123;</span><br><span class="line">      val label = bothV._3.get(<span class="string">"label"</span>).mkString</span><br><span class="line">      val expr = bothV._3.get(<span class="string">"expr"</span>).mkString</span><br><span class="line">      <span class="keyword">new</span> Tuple4[String, String, String, String](bothV._1.id().toString, bothV._2.id().toString, label, expr)</span><br><span class="line">    &#125;).toDF(<span class="string">"source"</span>, <span class="string">"target"</span>, <span class="string">"label"</span>, <span class="string">"expr"</span>)</span><br><span class="line">    bothVertexesDF.createOrReplaceTempView(<span class="string">"t_edge"</span>)</span><br><span class="line"></span><br><span class="line">    val categoryDF = labelsRdd.toDF(<span class="string">"category"</span>)</span><br><span class="line">    categoryDF.createOrReplaceTempView(<span class="string">"t_category"</span>)</span><br><span class="line"></span><br><span class="line">    val nodes = spark.sql(</span><br><span class="line">      <span class="string">""</span><span class="string">"</span></span><br><span class="line"><span class="string">        | select v.id, v.db, v.table, v.owner, v.tableType, v.col, v.colType, v.status, v.createAt, c.id - 1 category</span></span><br><span class="line"><span class="string">        |  from t_vertex v</span></span><br><span class="line"><span class="string">        |  left join</span></span><br><span class="line"><span class="string">        |  (</span></span><br><span class="line"><span class="string">        |   select row_number() over(order by category) id, category</span></span><br><span class="line"><span class="string">        |   from t_category</span></span><br><span class="line"><span class="string">        |  ) c on v.category = c.category</span></span><br><span class="line"><span class="string">        |</span></span><br><span class="line"><span class="string">      "</span><span class="string">""</span>.stripMargin)</span><br><span class="line">      .collect()</span><br><span class="line">      .map(r =&gt; &#123;</span><br><span class="line">        val id = r.getString(<span class="number">0</span>)</span><br><span class="line">        val db = r.getString(<span class="number">1</span>)</span><br><span class="line">        val table = r.getString(<span class="number">2</span>)</span><br><span class="line">        val owner = r.getString(<span class="number">3</span>)</span><br><span class="line">        val tableType = r.getString(<span class="number">4</span>)</span><br><span class="line">        val col = r.getString(<span class="number">5</span>)</span><br><span class="line">        val colType = r.getString(<span class="number">6</span>)</span><br><span class="line">        val status = r.getString(<span class="number">7</span>)</span><br><span class="line">        val createAt = r.getString(<span class="number">8</span>)</span><br><span class="line">        val category = r.getInt(<span class="number">9</span>)</span><br><span class="line">        val name = <span class="keyword">if</span> (StringUtils.isEmpty(col)) db + <span class="string">"."</span> + table <span class="keyword">else</span> db + <span class="string">"."</span> + table + <span class="string">"."</span> + col</span><br><span class="line">        val w = <span class="number">1000</span></span><br><span class="line">        val h = <span class="number">800</span></span><br><span class="line">        val x = Random.nextInt(w)</span><br><span class="line">        val y = Random.nextInt(h)</span><br><span class="line"></span><br><span class="line">        Vertex(id,</span><br><span class="line">          name,</span><br><span class="line">          db,</span><br><span class="line">          table,</span><br><span class="line">          owner,</span><br><span class="line">          tableType,</span><br><span class="line">          col,</span><br><span class="line">          colType,</span><br><span class="line">          status,</span><br><span class="line">          createAt,</span><br><span class="line">          category,</span><br><span class="line">          x,</span><br><span class="line">          y</span><br><span class="line">        )</span><br><span class="line">      &#125;)</span><br><span class="line"></span><br><span class="line">    val links = spark.sql(</span><br><span class="line">      <span class="string">""</span><span class="string">"</span></span><br><span class="line"><span class="string">        | select source, target, label, expr</span></span><br><span class="line"><span class="string">        | from t_edge</span></span><br><span class="line"><span class="string">      "</span><span class="string">""</span>.stripMargin)</span><br><span class="line">      .collect()</span><br><span class="line">      .map(r =&gt; &#123;</span><br><span class="line">        val source = r.getString(<span class="number">0</span>)</span><br><span class="line">        val target = r.getString(<span class="number">1</span>)</span><br><span class="line">        val label = r.getString(<span class="number">2</span>)</span><br><span class="line">        val expr = r.getString(<span class="number">3</span>)</span><br><span class="line"></span><br><span class="line">        Edge(source,</span><br><span class="line">          target,</span><br><span class="line">          label,</span><br><span class="line">          expr</span><br><span class="line">        )</span><br><span class="line">      &#125;)</span><br><span class="line"></span><br><span class="line">    val categories = spark.sql(</span><br><span class="line">      <span class="string">""</span><span class="string">"</span></span><br><span class="line"><span class="string">        | select category name</span></span><br><span class="line"><span class="string">        | from t_category</span></span><br><span class="line"><span class="string">        | order by category</span></span><br><span class="line"><span class="string">      "</span><span class="string">""</span>.stripMargin)</span><br><span class="line">      .collect()</span><br><span class="line">      .map(r =&gt; &#123;</span><br><span class="line">        val name = r.getString(<span class="number">0</span>)</span><br><span class="line"></span><br><span class="line">        Categories(name)</span><br><span class="line">      &#125;)</span><br><span class="line"></span><br><span class="line">    sc.parallelize(Array((nodes, links, categories)))</span><br><span class="line">      .toDF(<span class="string">"nodes"</span>, <span class="string">"links"</span>, <span class="string">"categories"</span>)</span><br><span class="line">      .createOrReplaceTempView(<span class="string">"t"</span>)</span><br><span class="line"></span><br><span class="line">    spark.sql(</span><br><span class="line">      <span class="string">""</span><span class="string">"</span></span><br><span class="line"><span class="string">        | select *</span></span><br><span class="line"><span class="string">        | from t</span></span><br><span class="line"><span class="string">      "</span><span class="string">""</span>.stripMargin)</span><br><span class="line">      .coalesce(<span class="number">1</span>)</span><br><span class="line">      .write.mode(SaveMode.Overwrite)</span><br><span class="line">      .json(<span class="string">"/graph"</span>)</span><br><span class="line"></span><br><span class="line">    spark.stop()</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">object GraphToJsonTask &#123;</span><br><span class="line">  <span class="function">def <span class="title">main</span><span class="params">(args: Array[String])</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">new</span> GraphToJsonService().save()</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="GraphDao"><a href="#GraphDao" class="headerlink" title="GraphDao"></a>GraphDao</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br><span class="line">76</span><br><span class="line">77</span><br><span class="line">78</span><br><span class="line">79</span><br><span class="line">80</span><br><span class="line">81</span><br><span class="line">82</span><br><span class="line">83</span><br><span class="line">84</span><br><span class="line">85</span><br><span class="line">86</span><br><span class="line">87</span><br><span class="line">88</span><br><span class="line">89</span><br><span class="line">90</span><br><span class="line">91</span><br><span class="line">92</span><br><span class="line">93</span><br><span class="line">94</span><br><span class="line">95</span><br><span class="line">96</span><br><span class="line">97</span><br><span class="line">98</span><br><span class="line">99</span><br><span class="line">100</span><br><span class="line">101</span><br><span class="line">102</span><br><span class="line">103</span><br><span class="line">104</span><br><span class="line">105</span><br><span class="line">106</span><br><span class="line">107</span><br><span class="line">108</span><br><span class="line">109</span><br><span class="line">110</span><br><span class="line">111</span><br><span class="line">112</span><br><span class="line">113</span><br><span class="line">114</span><br><span class="line">115</span><br><span class="line">116</span><br><span class="line">117</span><br><span class="line">118</span><br><span class="line">119</span><br><span class="line">120</span><br><span class="line">121</span><br><span class="line">122</span><br><span class="line">123</span><br><span class="line">124</span><br><span class="line">125</span><br><span class="line">126</span><br><span class="line">127</span><br><span class="line">128</span><br><span class="line">129</span><br><span class="line">130</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">package</span> spark.com.xyueji.traffic.dao</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> java.util</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.driver.Cluster</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.driver.remote.DriverRemoteConnection</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph</span><br><span class="line"><span class="keyword">import</span> org.apache.tinkerpop.gremlin.structure.&#123;Edge, Graph, Vertex&#125;</span><br><span class="line"></span><br><span class="line"><span class="keyword">import</span> scala.collection.JavaConverters._</span><br><span class="line"></span><br><span class="line"><span class="comment">/**</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@author</span> xiongzhigang</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@date</span> 2021-07-07 10:50</span></span><br><span class="line"><span class="comment">  * <span class="doctag">@description</span> </span></span><br><span class="line"><span class="comment">  */</span></span><br><span class="line"><span class="class"><span class="keyword">class</span> <span class="title">GraphDao</span> </span>&#123;</span><br><span class="line">  <span class="keyword">private</span> <span class="keyword">var</span> cluster: Cluster = _</span><br><span class="line">  <span class="keyword">private</span> <span class="keyword">var</span> graph: Graph = _</span><br><span class="line">  <span class="keyword">private</span> <span class="keyword">var</span> g: GraphTraversalSource = _</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">this</span><span class="params">(configurationFile: String, remoteTraversalSourceName: String)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">this</span>()</span><br><span class="line">    createConnection(configurationFile, remoteTraversalSourceName)</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="comment">/**</span></span><br><span class="line"><span class="comment">    * 创建连接</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">  <span class="function">def <span class="title">createConnection</span><span class="params">(configurationFile: String, remoteTraversalSourceName: String)</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="comment">//连接Gremlin Server</span></span><br><span class="line">    cluster = Cluster.open(configurationFile)</span><br><span class="line"></span><br><span class="line">    graph = EmptyGraph.instance</span><br><span class="line">    g = graph.traversal.withRemote(DriverRemoteConnection.using(cluster, remoteTraversalSourceName))</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">close</span><span class="params">()</span>: Unit </span>= &#123;</span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">      <span class="keyword">if</span> (g != <span class="keyword">null</span>) &#123; <span class="comment">// this closes the remote, no need to close the empty graph</span></span><br><span class="line">        g.close()</span><br><span class="line">      &#125;</span><br><span class="line">      <span class="keyword">if</span> (cluster != <span class="keyword">null</span>) &#123; <span class="comment">// the cluster closes all of its clients</span></span><br><span class="line">        cluster.close()</span><br><span class="line">      &#125;</span><br><span class="line">    &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">      g = <span class="keyword">null</span></span><br><span class="line">      graph = <span class="keyword">null</span></span><br><span class="line">      cluster = <span class="keyword">null</span></span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  def getGraph: GraphTraversalSource = g</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getV</span><span class="params">(id: AnyRef)</span>: Vertex </span>= &#123;</span><br><span class="line">    g.V(id).next()</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getV</span><span class="params">(key: String, value: String)</span>: Vertex </span>= &#123;</span><br><span class="line">    g.V().has(key, value).next()</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getProperties</span><span class="params">(id: AnyRef)</span>: Map[String, String] </span>= &#123;</span><br><span class="line">    val props = g.V(id).properties()</span><br><span class="line">    val label = getV(id).label()</span><br><span class="line"></span><br><span class="line">    val res = <span class="keyword">new</span> util.HashMap[String, String]()</span><br><span class="line">    res.put(<span class="string">"id"</span>, id.toString)</span><br><span class="line">    res.put(<span class="string">"label"</span>, label)</span><br><span class="line">    <span class="keyword">while</span> (props.hasNext) &#123;</span><br><span class="line">      val prop = props.next()</span><br><span class="line">      res.put(prop.key(), prop.value().toString)</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    res.asScala.toMap</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getEdgeProperties</span><span class="params">(id: AnyRef)</span>: Map[String, String] </span>= &#123;</span><br><span class="line">    val props = g.E(id).properties()</span><br><span class="line">    val label = getE(id).label()</span><br><span class="line"></span><br><span class="line">    val res = <span class="keyword">new</span> util.HashMap[String, String]()</span><br><span class="line">    res.put(<span class="string">"id"</span>, id.toString)</span><br><span class="line">    res.put(<span class="string">"label"</span>, label)</span><br><span class="line">    <span class="keyword">while</span> (props.hasNext) &#123;</span><br><span class="line">      val prop = props.next()</span><br><span class="line">      res.put(prop.key(), prop.value().toString)</span><br><span class="line">    &#125;</span><br><span class="line"></span><br><span class="line">    res.asScala.toMap</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">listAllVertexIds</span><span class="params">()</span>: util.List[AnyRef] </span>= &#123;</span><br><span class="line">    g.V().id().toList</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">listAllV</span><span class="params">()</span>: List[Map[String, String]] </span>= &#123;</span><br><span class="line">    val ids = listAllVertexIds()</span><br><span class="line">    val res = <span class="keyword">new</span> util.ArrayList[Map[String, String]]()</span><br><span class="line">    ids.asScala.foreach(id =&gt; &#123;</span><br><span class="line">      val map = getProperties(id)</span><br><span class="line">      res.add(map)</span><br><span class="line">    &#125;)</span><br><span class="line">    res.asScala.toList</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getE</span><span class="params">(id: AnyRef)</span>: Edge </span>= &#123;</span><br><span class="line">    g.E(id).next()</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  def listAllEdgeIds: util.List[AnyRef] = &#123;</span><br><span class="line">    g.E().id().toList</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="function">def <span class="title">getBothVertexByEdge</span><span class="params">(edge: Edge)</span>: <span class="params">(Vertex, Vertex, Map[String, String])</span> </span>= &#123;</span><br><span class="line">    (edge.outVertex(), edge.inVertex(), getEdgeProperties(edge.id()))</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  def listAllBothVertex: List[(Vertex, Vertex, Map[String, String])] = &#123;</span><br><span class="line">    val ids = listAllEdgeIds</span><br><span class="line">    val res = <span class="keyword">new</span> util.ArrayList[(Vertex, Vertex, Map[String, String])]()</span><br><span class="line">    ids.asScala.foreach(id =&gt; &#123;</span><br><span class="line">      val edge = getE(id)</span><br><span class="line">      res.add(getBothVertexByEdge(edge))</span><br><span class="line">    &#125;)</span><br><span class="line"></span><br><span class="line">    res.asScala.toList</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<blockquote>
<p>生成json文件后，Echarts就可以直接使用</p>
</blockquote>
<h4 id="Echarts文件"><a href="#Echarts文件" class="headerlink" title="Echarts文件"></a>Echarts文件</h4><p>参考：<a href="https://github.com/xyueji/traffic/blob/master/graph.html" target="_blank" rel="noopener">https://github.com/xyueji/traffic/blob/master/graph.html</a></p>
<h4 id="效果图-https-xyueji-github-io-traffic-graph-html"><a href="#效果图-https-xyueji-github-io-traffic-graph-html" class="headerlink" title="效果图(https://xyueji.github.io/traffic/graph.html):"></a><a href="https://xyueji.github.io/traffic/graph.html">效果图(https://xyueji.github.io/traffic/graph.html):</a></h4><p><img src="https://cdn.jsdelivr.net/gh/xyueji/img@master/web/image-20210902172517522.png" alt="image-20210902172517522"></p>
<p>参考：</p>
<ul>
<li><a href="https://dbaplus.cn/news-73-2143-1.html" target="_blank" rel="noopener">饿了么元数据管理实践之路</a></li>
<li><a href="https://docs.janusgraph.org/" target="_blank" rel="noopener">JanusGraph</a></li>
</ul>

      
      
        <div class="page-reward">
          <p><a href="javascript:void(0)" onclick="dashangToggle()" class="dashang">赏</a></p>
          <div class="hide_box"></div>
          <div class="shang_box">
            <a class="shang_close" href="javascript:void(0)" onclick="dashangToggle()">×</a>
            <div class="shang_tit">
              <p>纯属好玩</p>
            </div>
            <div class="shang_payimg">
              <img src="/img/alipayimg.jpg" alt="扫码支持" title="扫一扫" />
            </div>
              <div class="pay_explain">扫码打赏，你说多少就多少</div>
            <div class="shang_payselect">
              
                <div class="pay_item checked" data-id="alipay">
                  <span class="radiobox"></span>
                  <span class="pay_logo"><img src="/img/alipay.png" alt="支付宝" /></span>
                </div>
              
              
                <div class="pay_item" data-id="wechat">
                  <span class="radiobox"></span>
                  <span class="pay_logo"><img src="/img/weixin.png" alt="微信" /></span>
                </div>
              
            </div>
            <div class="shang_info">
              <p>打开<span id="shang_pay_txt">支付宝</span>扫一扫，即可进行扫码打赏哦</p>
            </div>
          </div>
        </div>
        <script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/zepto/1.2.0/zepto.min.js"></script>
        <script type="text/javascript">
          $(".pay_item").click(function(){
            $(this).addClass('checked').siblings('.pay_item').removeClass('checked');
            var dataid=$(this).attr('data-id');
            $(".shang_payimg img").attr("src","/img/"+dataid+"img.jpg");
            $("#shang_pay_txt").text(dataid=="alipay"?"支付宝":"微信");
          });
          function dashangToggle(){

            $(".hide_box").fadeToggle();
            $(".shang_box").fadeToggle();
          }
        </script>
      
    </div>
    
  </div>
  
    
    <div class="copyright">
        <p><span>本文标题:</span><a href="">元数据平台-数据血缘</a></p>
        <p><span>文章作者:</span><a href="/" title="访问 月 迹 的个人博客">月 迹</a></p>
        <p><span>发布时间:</span>2021年09月03日 - 17时39分</p>
        <p><span>最后更新:</span>2021年09月23日 - 17时17分</p>
        <p>
            <span>原始链接:</span><a class="post-url" href="" title="元数据平台-数据血缘">https://xyueji.github.io/元数据平台-数据血缘.html</a>
            <span class="copy-path" data-clipboard-text="原文: https://xyueji.github.io/元数据平台-数据血缘.html　　作者: 月 迹" title="点击复制文章链接"><i class="fa fa-clipboard"></i></span>
            <script src="/js/clipboard.min.js"></script>
            <script> var clipboard = new Clipboard('.copy-path'); </script>
        </p>
        <p>
            <span>许可协议:</span><i class="fa fa-creative-commons"></i> <a rel="license" href="http://creativecommons.org/licenses/by-nc-sa/3.0/cn/" title="中国大陆 (CC BY-NC-SA 3.0 CN)" target = "_blank">"署名-非商用-相同方式共享 3.0"</a> 转载请保留原文链接及作者。
        </p>
    </div>



<nav id="article-nav">
  
    <a href="统计HDFS元数据.html" id="article-nav-newer" class="article-nav-link-wrap">
      <strong class="article-nav-caption"><</strong>
      <div class="article-nav-title">
        
          统计HDFS元数据
        
      </div>
    </a>
  
  
    <a href="精准去重-字典编码+Bitmap.html" id="article-nav-older" class="article-nav-link-wrap">
      <div class="article-nav-title">精准去重-字典编码+Bitmap</div>
      <strong class="article-nav-caption">></strong>
    </a>
  
</nav>


  
</article>

    <div id="toc" class="toc-article">
    <strong class="toc-title">文章目录</strong>
    <ol class="toc"><li class="toc-item toc-level-3"><a class="toc-link" href="#架构"><span class="toc-number">1.</span> <span class="toc-text">架构</span></a></li><li class="toc-item toc-level-3"><a class="toc-link" href="#SQL采集"><span class="toc-number">2.</span> <span class="toc-text">SQL采集</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#SqlCollectionHook"><span class="toc-number">2.1.</span> <span class="toc-text">SqlCollectionHook</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#SqlCollectionStrategy"><span class="toc-number">2.2.</span> <span class="toc-text">SqlCollectionStrategy</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#KafkaSqlCollectionStrategy"><span class="toc-number">2.3.</span> <span class="toc-text">KafkaSqlCollectionStrategy</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#JdbcSqlCollectionStrategy"><span class="toc-number">2.4.</span> <span class="toc-text">JdbcSqlCollectionStrategy</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#hive-site-xml"><span class="toc-number">2.5.</span> <span class="toc-text">hive-site.xml</span></a></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#SQL解析"><span class="toc-number">3.</span> <span class="toc-text">SQL解析</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#SQLParse2"><span class="toc-number">3.1.</span> <span class="toc-text">SQLParse2</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#Flink任务"><span class="toc-number">3.2.</span> <span class="toc-text">Flink任务</span></a><ol class="toc-child"><li class="toc-item toc-level-5"><a class="toc-link" href="#LineageTask"><span class="toc-number">3.2.1.</span> <span class="toc-text">LineageTask</span></a></li><li class="toc-item toc-level-5"><a class="toc-link" href="#SqlEntity"><span class="toc-number">3.2.2.</span> <span class="toc-text">SqlEntity</span></a></li><li class="toc-item toc-level-5"><a class="toc-link" href="#LineageProcessWindowFunction"><span class="toc-number">3.2.3.</span> <span class="toc-text">LineageProcessWindowFunction</span></a></li><li class="toc-item toc-level-5"><a class="toc-link" href="#LineageBuildMapFunction"><span class="toc-number">3.2.4.</span> <span class="toc-text">LineageBuildMapFunction</span></a></li><li class="toc-item toc-level-5"><a class="toc-link" href="#LineageToGraphSinkFunction"><span class="toc-number">3.2.5.</span> <span class="toc-text">LineageToGraphSinkFunction</span></a></li><li class="toc-item toc-level-5"><a class="toc-link" href="#remote-yaml"><span class="toc-number">3.2.6.</span> <span class="toc-text">remote.yaml</span></a></li></ol></li></ol></li><li class="toc-item toc-level-3"><a class="toc-link" href="#数据血缘查询"><span class="toc-number">4.</span> <span class="toc-text">数据血缘查询</span></a><ol class="toc-child"><li class="toc-item toc-level-4"><a class="toc-link" href="#GraphToJsonTask"><span class="toc-number">4.1.</span> <span class="toc-text">GraphToJsonTask</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#GraphDao"><span class="toc-number">4.2.</span> <span class="toc-text">GraphDao</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#Echarts文件"><span class="toc-number">4.3.</span> <span class="toc-text">Echarts文件</span></a></li><li class="toc-item toc-level-4"><a class="toc-link" href="#效果图-https-xyueji-github-io-traffic-graph-html"><span class="toc-number">4.4.</span> <span class="toc-text">效果图(https://xyueji.github.io/traffic/graph.html):</span></a></li></ol></li></ol>
</div>
<input type="button" id="tocButton" value="隐藏目录"  title="点击按钮隐藏或者显示文章目录">

<script src="https://cdnjs.cloudflare.com/ajax/libs/require.js/2.1.6/require.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/1.9.1/jquery.min.js"></script>
<script>
    var valueHide = "隐藏目录";
    var valueShow = "显示目录";

    if ($(".left-col").is(":hidden")) {
        $("#tocButton").attr("value", valueShow);
    }
    $("#tocButton").click(function() {
        if ($("#toc").is(":hidden")) {
            $("#tocButton").attr("value", valueHide);
            $("#toc").slideDown(320);
        }
        else {
            $("#tocButton").attr("value", valueShow);
            $("#toc").slideUp(350);
        }
    })
    if ($(".toc").length < 1) {
        $("#toc, #tocButton").hide();
    }
</script>






<div class="bdsharebuttonbox">
	<a href="#" class="fx fa-weibo bds_tsina" data-cmd="tsina" title="分享到新浪微博"></a>
	<a href="#" class="fx fa-weixin bds_weixin" data-cmd="weixin" title="分享到微信"></a>
	<a href="#" class="fx fa-qq bds_sqq" data-cmd="sqq" title="分享到QQ好友"></a>
	<a href="#" class="fx fa-facebook-official bds_fbook" data-cmd="fbook" title="分享到Facebook"></a>
	<a href="#" class="fx fa-twitter bds_twi" data-cmd="twi" title="分享到Twitter"></a>
	<a href="#" class="fx fa-linkedin bds_linkedin" data-cmd="linkedin" title="分享到linkedin"></a>
	<a href="#" class="fx fa-files-o bds_copy" data-cmd="copy" title="分享到复制网址"></a>
</div>
<script>window._bd_share_config={"common":{"bdSnsKey":{},"bdText":"","bdMini":"2","bdMiniList":false,"bdPic":"","bdStyle":"2","bdSize":"24"},"share":{}};with(document)0[(getElementsByTagName('head')[0]||body).appendChild(createElement('script')).src='/static/api/js/share.js?v=89860593.js?cdnversion='+~(-new Date()/36e5)];</script>




    
        <link rel="stylesheet" href="https://cdn.bootcdn.net/ajax/libs/gitalk/1.6.2/gitalk.min.css">
<div id="gitalk-container" class="gitment-container"></div>
<script type="text/javascript">
    var gitalk = new Gitalk({
        clientID: '8864436c2b03c3753a80',
        clientSecret: '252abc963e694743b034294b00a36ec6c7ee2152',
        id: '元数据平台-数据血缘.html',
        owner: 'xyueji',
        repo: 'xyueji.github.io',
        admin: 'xyueji',
        proxy: 'https://still-shadow-c49f.xyueji.workers.dev/?https://github.com/login/oauth/access_token'
    })
    gitalk.render('gitalk-container')
</script>

    



    <div class="scroll" id="post-nav-button">
        
            <a href="统计HDFS元数据.html" title="上一篇: 统计HDFS元数据">
                <i class="fa fa-angle-left"></i>
            </a>
        
        <a title="文章列表"><i class="fa fa-bars"></i><i class="fa fa-times"></i></a>
        
            <a href="精准去重-字典编码+Bitmap.html" title="下一篇: 精准去重-字典编码+Bitmap">
                <i class="fa fa-angle-right"></i>
            </a>
        
    </div>
    <ul class="post-list"><li class="post-list-item"><a class="post-list-link" href="统计HDFS元数据.html">统计HDFS元数据</a></li><li class="post-list-item"><a class="post-list-link" href="">元数据平台-数据血缘</a></li><li class="post-list-item"><a class="post-list-link" href="精准去重-字典编码+Bitmap.html">精准去重-字典编码+Bitmap</a></li><li class="post-list-item"><a class="post-list-link" href="大数据监控-Metrics2录入指标到OpenTSDB.html">大数据监控-Metrics2录入指标到OpenTSDB</a></li><li class="post-list-item"><a class="post-list-link" href="Spark解决数据倾斜问题.html">Spark解决数据倾斜问题</a></li><li class="post-list-item"><a class="post-list-link" href="Spark-Shuffle原理.html">Spark-Shuffle原理</a></li><li class="post-list-item"><a class="post-list-link" href="MapReduce-Shuffle过程.html">MapReduce-Shuffle过程</a></li><li class="post-list-item"><a class="post-list-link" href="Git开发规范.html">Git开发规范</a></li><li class="post-list-item"><a class="post-list-link" href="Flink学习--如何计算实时热门商品.html">Flink学习--如何计算实时热门商品</a></li><li class="post-list-item"><a class="post-list-link" href="Flink学习--基本概念与环境搭建.html">Flink学习--基本概念与环境搭建</a></li><li class="post-list-item"><a class="post-list-link" href="docker-compose搭建大数据环境.html">docker-compose搭建大数据环境</a></li><li class="post-list-item"><a class="post-list-link" href="Mac-docker动态给Container添加端口映射.html">Mac-docker动态给Container添加端口映射</a></li><li class="post-list-item"><a class="post-list-link" href="Vue+SpringBoot项目docker一键部署.html">Vue+SpringBoot项目docker一键部署</a></li><li class="post-list-item"><a class="post-list-link" href="vue-simple-uploader+springboot大文件上传.html">vue-simple-uploader+springboot大文件上传</a></li><li class="post-list-item"><a class="post-list-link" href="IDEA连接Impala.html">IDEA连接Impala</a></li><li class="post-list-item"><a class="post-list-link" href="HBase基本知识.html">HBase基本知识</a></li><li class="post-list-item"><a class="post-list-link" href="IDE破解.html">IDE破解</a></li><li class="post-list-item"><a class="post-list-link" href="salt-api配置.html">salt-api配置</a></li><li class="post-list-item"><a class="post-list-link" href="线上HBase表重建.html">线上HBase表重建</a></li><li class="post-list-item"><a class="post-list-link" href="Alluxio+Presto多版本测试.html">Alluxio+Presto多版本测试</a></li><li class="post-list-item"><a class="post-list-link" href="Linux-top命令详解.html">Linux-top命令详解</a></li><li class="post-list-item"><a class="post-list-link" href="Linux-free命令详解.html">Linux-free命令详解</a></li><li class="post-list-item"><a class="post-list-link" href="hbase的hbck工具.html">hbase的hbck工具</a></li><li class="post-list-item"><a class="post-list-link" href="JanusGraph.html">JanusGraph</a></li><li class="post-list-item"><a class="post-list-link" href="iptables禁用启用端口.html">iptables禁用启用端口</a></li><li class="post-list-item"><a class="post-list-link" href="Phoenix系统表.html">Phoenix系统表</a></li><li class="post-list-item"><a class="post-list-link" href="Phoenix安装问题.html">Phoenix安装问题</a></li><li class="post-list-item"><a class="post-list-link" href="Prometheus安装-HDFS指标收集.html">Prometheus安装+HDFS指标收集</a></li></ul>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/require.js/2.1.6/require.js">
    <script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/1.9.1/jquery.min.js">
    <script>
        $(".post-list").addClass("toc-article");
        $(".post-list-item a").attr("target","_blank");
        $("#post-nav-button > a:nth-child(2)").click(function() {
            $(".fa-bars, .fa-times").toggle();
            $(".post-list").toggle(300);
            if ($(".toc").length > 0) {
                $("#toc, #tocButton").toggle(200, function() {
                    if ($(".switch-area").is(":visible")) {
                        $("#tocButton").attr("value", valueHide);
                        }
                    })
            }
            else {
            }
        })
    </script>




    <script>
        
    </script>
   <!-- 代码块复制功能-->
   <script src="/js/clipboard-use.js"></script>

</div>
      <footer id="footer">
    <div class="outer">
        <div id="footer-info">
            <div class="footer-left">
                &copy; 2022 月 迹
            </div>
            <div class="footer-right">
                <a href="http://hexo.io/" target="_blank" style="color: #000">Hexo</a>  Theme <a href="https://github.com/luuman/hexo-theme-spfk" target="_blank" style="color: #000">spfk</a> by luuman
            </div>
        </div>
        
            <div class="visit">
                
                    <span id="busuanzi_container_site_pv" style='display:none'>
                        <span id="site-visit" >本站到访数:
                            <span id="busuanzi_value_site_uv"></span>
                        </span>
                    </span>
                
                
                    <span>, </span>
                
                
                    <span id="busuanzi_container_page_pv" style='display:none'>
                        <span id="page-visit">本页阅读量:
                            <span id="busuanzi_value_page_pv"></span>
                        </span>
                    </span>
                
            </div>
        
    </div>
</footer>

    </div>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/require.js/2.1.6/require.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/1.9.1/jquery.min.js"></script>
<script src="js/main.js"></script>

    <!--<script>-->
        <!--$(document).ready(function() {-->
            <!--var backgroundnum = 0;-->
            <!--var backgroundimg = "url(/background/bg-" + backgroundnum +".jpg)";-->
            <!--$("#mobile-nav").css({"background-image": backgroundimg,"background-size": "cover","background-position": "center"});-->
            <!--$(".left-col").css({"background-image": backgroundimg,"background-size": "cover","background-position": "center"});-->
        <!--})-->
    <!--</script>-->



<!-- Google Analytics -->
<!-- Global site tag (gtag.js) - Google Analytics -->
<script async src="https://www.googletagmanager.com/gtag/js?id=UA-168200248-1"></script>
<script>
    window.dataLayer = window.dataLayer || [];
    function gtag(){dataLayer.push(arguments);}
    gtag('js', new Date());

    gtag('config', 'UA-168200248-1');
</script>

<!-- End Google Analytics -->



<script>
	var _hmt = _hmt || [];
	(function() {
		var hm = document.createElement("script");
		hm.src = "https://hm.baidu.com/hm.js?1c0d5d76b7455fc584663ce315f75837";
		var s = document.getElementsByTagName("script")[0];
		s.parentNode.insertBefore(hm, s);
	})();
</script>



<div class="scroll" id="scroll">
    <a href="#"><i class="fa fa-arrow-up"></i></a>
    <a href="#comments"><i class="fa fa-comments-o"></i></a>
    <a href="#footer"><i class="fa fa-arrow-down"></i></a>
</div>
<script>
    $(document).ready(function() {
        if ($("#comments").length < 1) {
            $("#scroll > a:nth-child(2)").hide();
        };
    })
</script>


<script async src="//busuanzi.ibruce.info/busuanzi/2.3/busuanzi.pure.mini.js"></script>
    
    <script language="javascript">
    $(function() {
        $("a[title]").each(function() {
            var a = $(this);
            var title = a.attr('title');
            if (title == undefined || title == "") return;
            a.data('title', title).removeAttr('title').hover(

            function() {
                var offset = a.offset();
                $("<div id=\"anchortitlecontainer\"></div>").appendTo($("body")).html(title).css({
                    top: offset.top - a.outerHeight() - 15,
                    left: offset.left + a.outerWidth()/2 + 1
                }).fadeIn(function() {
                    var pop = $(this);
                    setTimeout(function() {
                        pop.remove();
                    }, pop.text().length * 800);
                });
            }, function() {
                $("#anchortitlecontainer").remove();
            });
        });
    });
</script>
    
    
    
        <div class="search-result-list" ></div>
        <div class="search-cover"></div>
    <script type = "text/javascript" >
        $('#search').bind('keypress',function(event){
            if(event.keyCode == "13"){
                search()
            }
        });

    function search() {
        var searchTool = {
            data: [],
            arrList: [],
            oldList: [],
            init() {
                this.fetchJson()
            },
            fetchJson() {
                window.fetch('/content.json?t=' + (+new Date()), {
                    method: 'get',
                }).then((res) => {
                    return res.json()
                }).then((data) => {
                    this.data = data
                    let keywords = $('#search').val()
                    this.searchList(keywords)
                }).catch((err) => {
                    console.log(err)
                });
            },
            // 搜索字符串里面是否存在关键字
            isSreachIndexOF(oldstr, kw) {
                // console.log(oldstr, kw)
                var istrue = false
                // console.log('isSreachIndexOF', oldstr && toString.call(oldstr) === '[object Array]')
                if (oldstr && toString.call(oldstr) === '[object Array]') {
                    for (var i = 0; i < oldstr.length; i++) {
                        oldstr[i].toLowerCase() === kw.toLowerCase() ? istrue = true : null
                    }
                    return istrue
                }
                if (!oldstr || !kw) return false
                istrue = oldstr.toLowerCase().indexOf(kw.toLowerCase()) > -1
                return istrue
            },
            searchList(keywolds) {
                let divList = ''
                if (!keywolds) {
                    if (this.oldList.length) this.clear()
                } else {
                    this.data.forEach((item, index) => {
                        let tagsK = item.tags.map(item => item.name).join('-')
                        let categoriesK = item.categories.map(item => item.name).join('-')
                        if (this.isSreachIndexOF(item.title, keywolds) || this.isSreachIndexOF(tagsK, keywolds) || this.isSreachIndexOF(categoriesK, keywolds) || this.isSreachIndexOF(item.content, keywolds)) {
                            divList += `
                          <li>
                            <a href="/${item.path}" class="search-result-title">${item.title}</a>
                            <div class="main">
                              <div class="left">
                                <span>${item.author ? item.author : 'xyueji'}</span>
                                ${item.tags.map(tags => `<a href="/tags/${tags.slug}" class="search-result-tags">${tags.name}</a>`)}
                              </div>
                              <div class="right">${this.getFriendlyTime(item.date.replace('T', ' '), new Date())}</div>
                            </div>
                          </li>`
                        }
                    })
                    if (this.oldList.length) {
                        this.clear()
                    }
                    if (this.arrList.length) this.add()
                }
                if (!divList) {
                    divList = `<a class="search_close" href="javascript:void(0)" onclick="searchToggle()">×</a> <div style="text-align: center;font-size: 30px; margin-top: 50px;" ">抱歉，未搜索到相关结果</div>`
                } else {
                    divList = `<a class="search_close" href="javascript:void(0)" onclick="searchToggle()">×</a> <ul>` + divList + `</ul>`
                }
                $('.search-result-list').html(divList)
                $('.search-result-list').fadeToggle();
                $('.search-cover').fadeToggle();
            },
            clear() {
                this.oldList.forEach(item => {
                    $(".search-result-list li")[item].style.display = 'none'
                })
                this.oldList = []
            },
            add() {
                this.arrList.forEach(item => {
                    $(".search-result-list li")[item].style.display = 'block'
                })
                this.oldList = this.arrList
            },
            searchGo(keywolds) {
                this.arrList = []
                if (!keywolds) {
                    if (this.oldList.length) this.clear()
                } else {
                    this.data.forEach((item, index) => {
                        if (this.isSreachIndexOF(item.title, keywolds) || this.isSreachIndexOF(item.tags.map(item => item.name).join('-'), keywolds) || this.isSreachIndexOF(item.categories.map(item => item.name).join('-'), keywolds) || this.isSreachIndexOF(item.content, keywolds)) {
                            this.arrList.push(index)
                        }
                    })
                    if (this.oldList.length) this.clear()
                    if (this.arrList.length) this.add()
                }
            },
            /**
             * 获取指定时间的友好时间字符串。
             * @param str 指定的时间字符串，如yyyy-MM-dd HH:mm:ss
             * @param now 当前时间，允许时间戳，GMT时间，如果该参数为undefined，则使用浏览器时间。
             */
            getFriendlyTime(str, now) {
                var currentTime = new Date(now);
                var arr = str.split(/\s+/gi);
                var temp = 0, arr1, arr2, oldTime, delta;
                var getIntValue = function (ss, defaultValue) {
                    try {
                        return parseInt(ss, 10);
                    } catch (e) {
                        return defaultValue;
                    }
                };
                var getWidthString = function (num) {
                    return num < 10 ? ('0' + num) : num;
                };
                if (arr.length >= 2) {
                    arr1 = arr[0].split(/[\/\-]/gi);
                    arr2 = arr[1].split(':');
                    oldTime = new Date();
                    oldTime.setYear(getIntValue(arr1[0], currentTime.getFullYear()));
                    oldTime.setMonth(getIntValue(arr1[1], currentTime.getMonth() + 1) - 1);
                    oldTime.setDate(getIntValue(arr1[2], currentTime.getDate()));

                    oldTime.setHours(getIntValue(arr2[0], currentTime.getHours()));
                    oldTime.setMinutes(getIntValue(arr2[1], currentTime.getMinutes()));
                    oldTime.setSeconds(getIntValue(arr2[2], currentTime.getSeconds()));

                    delta = currentTime.getTime() - oldTime.getTime();

                    if (delta <= 60 * 1000) {
                        return '1分钟内';
                    } else if (delta < 60 * 60 * 1000) {
                        return Math.floor(delta / (60 * 1000)) + '分钟前';
                    } else if (delta < 24 * 60 * 60 * 1000) {
                        return Math.floor(delta / (60 * 60 * 1000)) + '小时前';
                    } else if (delta < 24 * 60 * 60 * 1000 * 30) {
                        return Math.floor(delta / (24 * 60 * 60 * 1000)) + '天前';
                    } else if (delta < 24 * 60 * 60 * 1000 * 30 * 12) {
                        return Math.floor(delta / (24 * 60 * 60 * 1000 * 30)) + '月前';
                    } else if (currentTime.getFullYear() != oldTime.getFullYear()) {
                        return [getWidthString(oldTime.getFullYear()), getWidthString(oldTime.getMonth() + 1), getWidthString(oldTime.getDate())].join('-')
                    } else {
                        return [getWidthString(oldTime.getMonth() + 1), getWidthString(oldTime.getDate())].join('-');
                    }
                }
                return '';
            }
        }
        searchTool.init()
    }
    function searchToggle() {
        $('.search-cover').fadeToggle();
        $('.search-result-list').fadeToggle();
    }
</script>


  </div>
</body>
</html>
